package com.o2o.regularservice.dws_orc

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.lit
import org.elasticsearch.spark._
object ES_ORC_OBS {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .config("es.nodes", "192.168.1.29")
      .config("es.port", "9200")
      .config("cluster.name", "O2OElastic")
      .config("es.net.http.auth.user", "elastic")
      .config("es.net.http.auth.pass", "changeme")
      .config("es.index.read.missing.as.empty", "true")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.hadoop.fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
      .config("spark.hadoop.fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
      .config("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
      .getOrCreate()
    val sc = spark.sparkContext
    sc.setLogLevel("WARN")
    //年份
    val year = "2021"
    //月份
    var arrMonth = (2 to 2).toArray
//    var platArr = Array( "tmall","jd","meituan","elm","suning","pinduoduo","meituan_tg","dzdp","rongyigou","beibeiapp","qichezhijia",
//                          "hmxs","miya","kaola","secoo","womaiwang","yunji","dangdang",
//                          "guomei","jumei","tangeche","mryx")

    var platArr = Array("rongyigou","beibeiapp","qichezhijia","miya","kaola","secoo","womaiwang","yunji","dangdang","guomei","jumei","tangeche","beidian","jiuxian","mryx")

    //"rongyigou","beibeiapp","qichezhijia","kaola","secoo","womaiwang","yunji","dangdang",
//    "guomei","jumei","tangeche","mryx","beidian","jiuxian"
//    "miya",


//    ,"jd_daojia","txd"


//,"mryx","beidian","elm"
//    var platArr = Array("beidian","jiuxian")
//    var platArr = Array("rongyigou")
//    var platArr = Array("tmall","jd","meituan","elm","suning","pinduoduo","meituan_tg","dzdp","txd")

    //新增的平台
//    var platArr = Array("mryx")

    for (platform <- platArr) {
      //=======四个政府平台  苏宁、拼多多、美团团购和大众点评
      if(platform =="tmall"){
        ES_ORC_TMALL(spark,year,platform,arrMonth)
      }else if(platform =="jd"){
        ES_ORC_JD(spark,year,platform,arrMonth)
      }else if(platform =="meituan"){
        ES_ORC_MT(spark,year,platform,arrMonth)
      }else if(platform =="elm"){
        ES_ORC_ELM(spark,year,platform,arrMonth)
      }else if(platform =="suning"){
        ES_ORC_SUNING(spark,year,platform,arrMonth)
      }
      else if(platform =="pinduoduo"){
        ES_ORC_PDD(spark,year,platform,arrMonth)
      }
      else if(platform =="meituan_tg"){
        ES_ORC_MTTG(spark,year,platform,arrMonth)
      } else if(platform =="dzdp"){
        ES_ORC_DZDP(spark,year,platform,arrMonth)
      }
      //======16个小平台
      else if(platform =="jiuxian"){
        ES_ORC_JX(spark,year,platform,arrMonth)
      }
      else if(platform =="jd_daojia"){
        ES_ORC_JDDJ(spark,year,platform,arrMonth)
      }
      else if(platform =="beidian"){
        ES_ORC_BeiDian(spark,year,platform,arrMonth)
      }
      else if(platform =="rongyigou"){
        ES_ORC_RYG(spark,year,platform,arrMonth)
      }
      else if(platform =="beibeiapp"){
        ES_OBS_2_beibei(spark,year,platform,arrMonth)
      }
      else if(platform =="qichezhijia"){
        ES_OBS_3_qichezhijia(spark,year,platform,arrMonth)
      }
      else if(platform =="babytree"){
        ES_OBS_4_babytree(spark,year,platform,arrMonth)
      }
      else if(platform =="mryx"){
        ES_OBS_5_mryx(spark,year,platform,arrMonth)
      }
      else if(platform =="txd"){
        ES_OBS_6_hmxs(spark,year,platform,arrMonth)
      }
      else if(platform =="shunfeng"){
        ES_OBS_7_shunfeng(spark,year,platform,arrMonth)
      }
      else if(platform =="miya"){
        ES_OBS_8_miya(spark,year,platform,arrMonth)
      }
      else if(platform =="kaola"){
        ES_OBS_9_kaola(spark,year,platform,arrMonth)
      }
      else if(platform =="secoo"){
        ES_OBS_10_secoo(spark,year,platform,arrMonth)
      }
      else if(platform =="womaiwang"){
        ES_OBS_11_womaiwang(spark,year,platform,arrMonth)
      }
      else if(platform =="yunji"){
        ES_OBS_12_yunji(spark,year,platform,arrMonth)
      }
      else if(platform =="dangdang"){
        ES_OBS_13_dangdang(spark,year,platform,arrMonth)
      }
      else if(platform =="guomei"){
        ES_OBS_14_guomei(spark,year,platform,arrMonth)
      }
      else if(platform =="jumei"){
        ES_OBS_15_jumei(spark,year,platform,arrMonth)
      }
      else if(platform =="tangeche"){
        ES_OBS_16_tangeche(spark,year,platform,arrMonth)
      }
    }
  }
//==============================政府平台提取方法======================================================//

  //天猫
  def ES_ORC_TMALL(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {
      val frame = spark.read.orc(s"s3a://dws-data/g_data/${year}/${month}/tmall/*")
      if(frame != null){
        val value = frame.toJSON.rdd.map(line => {
          val nObject = JSON.parseObject(line)
          val BC_type = nObject.getOrDefault("BC_type","-1").toString
          val Base_Info = nObject.getOrDefault("Base_Info","-1").toString
          val allItemCount = nObject.getOrDefault("allItemCount","-1").toString
          val brandName = nObject.getOrDefault("brandName","-1").toString
          val brandName_cn = nObject.getOrDefault("brandName_cn","-1").toString
          val brandName_en = nObject.getOrDefault("brandName_en","-1").toString
          val brandSourceId = nObject.getOrDefault("brandSourceId","-1").toString
          val brandValueId = nObject.getOrDefault("brandValueId","-1").toString
          val brand_isLaoZiHao = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val brand_state = nObject.getOrDefault("brand_state","-1").toString
          val brand_type = nObject.getOrDefault("brand_type","-1").toString
          val categoryId = nObject.getOrDefault("categoryId","-1").toString
          val commentCount = nObject.getOrDefault("commentCount","-1").toString
          val creditLevel = nObject.getOrDefault("creditLevel","-1").toString
          val delivery_from = nObject.getOrDefault("delivery_from","-1").toString
          val detailCategoryName = nObject.getOrDefault("detailCategoryName","-1").toString
          val evaluates = nObject.getOrDefault("evaluates","-1").toString
          val fans = nObject.getOrDefault("fans","-1").toString
          val favcount = nObject.getOrDefault("favcount","-1").toString
          val firstCategoryId = nObject.getOrDefault("firstCategoryId","-1").toString
          val fourthCategoryId = nObject.getOrDefault("fourthCategoryId","-1").toString
          val goodRatePercentage = nObject.getOrDefault("goodRatePercentage","-1").toString
          val goodUrl = nObject.getOrDefault("goodUrl","-1").toString
          val good_id = nObject.getOrDefault("good_id","-1").toString
          val images = nObject.getOrDefault("images","-1").toString
          val is_flagship = nObject.getOrDefault("is_flagship","-1").toString
          val is_market = nObject.getOrDefault("is_market","-1").toString
          val is_o2o = nObject.getOrDefault("is_o2o","-1").toString
          val is_onsell = nObject.getOrDefault("is_onsell","-1").toString
          val is_oversea = nObject.getOrDefault("is_oversea","-1").toString
          val originCountryName = nObject.getOrDefault("originCountryName","-1").toString
          val originItem = nObject.getOrDefault("originItem","-1").toString
          val originProduction = nObject.getOrDefault("originProduction","-1").toString
          val original_cost = nObject.getOrDefault("original_cost","-1").toString
          val platformId = nObject.getOrDefault("platformId","-1").toString
          val platformName = nObject.getOrDefault("platformName","-1").toString
          val priceText = nObject.getOrDefault("priceText","-1").toString
          val products = nObject.getOrDefault("products","-1").toString
          val quantity = nObject.getOrDefault("quantity","-1").toString
          val rootCategoryId = nObject.getOrDefault("rootCategoryId","-1").toString
          val salesAmount = nObject.getOrDefault("salesAmount","-1").toString
          val secondCategoryId = nObject.getOrDefault("secondCategoryId","-1").toString
          val sellCount = nObject.getOrDefault("sellCount","-1").toString
          val sellerNick = nObject.getOrDefault("sellerNick","-1").toString
          val sellerType = nObject.getOrDefault("sellerType","-1").toString
          val shopId = nObject.getOrDefault("shopId","-1").toString
          val shopName = nObject.getOrDefault("shopName","-1").toString
          val shopType = nObject.getOrDefault("shopType","-1").toString
          val shopUrl = nObject.getOrDefault("shopUrl","-1").toString
          val starts = nObject.getOrDefault("starts","-1").toString
          val swb_products = nObject.getOrDefault("swb_products","-1").toString
          val swbfirstId = nObject.getOrDefault("swbfirstId","-1").toString
          val swbfirstName = nObject.getOrDefault("swbfirstName","-1").toString
          val taxFreeZone = nObject.getOrDefault("taxFreeZone","-1").toString
          val thirdCategoryId = nObject.getOrDefault("thirdCategoryId","-1").toString
          val timeStamp = nObject.getOrDefault("timeStamp","-1").toString
          val title = nObject.getOrDefault("title","-1").toString
          val unite_brand = nObject.getOrDefault("unite_brand","-1").toString
          val userId = nObject.getOrDefault("userId","-1").toString

          val  address =  nObject.getOrDefault("address","-1").toString
          val  administrative_region =  nObject.getOrDefault("administrative_region","-1").toString
          val  aedzId =  nObject.getOrDefault("aedzId","-1").toString
          val  city =  nObject.getOrDefault("city","-1").toString
          val  city_grade =  nObject.getOrDefault("city_grade","-1").toString
          val  city_origin =  nObject.getOrDefault("city_origin","-1").toString
          val  district =  nObject.getOrDefault("district","-1").toString
          val  district_origin =  nObject.getOrDefault("district_origin","-1").toString
          val  economic_division =  nObject.getOrDefault("economic_division","-1").toString
          val  if_city =  nObject.getOrDefault("if_city","-1").toString
          val  if_district =  nObject.getOrDefault("if_district","-1").toString
          val  if_state_level_new_areas =  nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val  latitude =  nObject.getOrDefault("latitude","-1").toString
          val  longitude =  nObject.getOrDefault("longitude","-1").toString
          val  name =  nObject.getOrDefault("name","-1").toString
          val  poor_counties =  nObject.getOrDefault("poor_counties","-1").toString
          val  province =  nObject.getOrDefault("province","-1").toString
          val  regional_ID =  nObject.getOrDefault("regional_ID","-1").toString
          val  registration_institution =  nObject.getOrDefault("registration_institution","-1").toString
          val  rural_demonstration_counties =  nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val  rural_ecommerce =  nObject.getOrDefault("rural_ecommerce","-1").toString
          val  the_belt_and_road_city =  nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val  the_belt_and_road_province =  nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val  the_yangtze_river_economic_zone_city =  nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val  the_yangtze_river_economic_zone_province =  nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val  town =  nObject.getOrDefault("town","-1").toString
          val  urban_agglomerations =  nObject.getOrDefault("urban_agglomerations","-1").toString

          nObject.put("BC_type",BC_type)
          nObject.put("Base_Info",Base_Info)
          nObject.put("allItemCount",allItemCount)
          nObject.put("brandName",brandName)
          nObject.put("brandName_cn",brandName_cn)
          nObject.put("brandName_en",brandName_en)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("brandValueId",brandValueId)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("brand_state",brand_state)
          nObject.put("brand_type",brand_type)
          nObject.put("categoryId",categoryId)
          nObject.put("commentCount",commentCount)
          nObject.put("creditLevel",creditLevel)
          nObject.put("delivery_from",delivery_from)
          nObject.put("detailCategoryName",detailCategoryName)
          nObject.put("evaluates",evaluates)
          nObject.put("fans",fans)
          nObject.put("favcount",favcount)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("goodRatePercentage",goodRatePercentage)
          nObject.put("goodUrl",goodUrl)
          nObject.put("good_id",good_id)
          nObject.put("images",images)
          nObject.put("is_flagship",is_flagship)
          nObject.put("is_market",is_market)
          nObject.put("is_o2o",is_o2o)
          nObject.put("is_onsell",is_onsell)
          nObject.put("is_oversea",is_oversea)
          nObject.put("originCountryName",originCountryName)
          nObject.put("originItem",originItem)
          nObject.put("originProduction",originProduction)
          nObject.put("original_cost",original_cost)
          nObject.put("platformId",platformId)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("products",products)
          nObject.put("quantity",quantity)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("salesAmount",salesAmount)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("sellCount",sellCount)
          nObject.put("sellerNick",sellerNick)
          nObject.put("sellerType",sellerType)
          nObject.put("shopId",shopId)
          nObject.put("shopName",shopName)
          nObject.put("shopType",shopType)
          nObject.put("shopUrl",shopUrl)
          nObject.put("starts",starts)
          nObject.put("swb_products",swb_products)
          nObject.put("swbfirstId",swbfirstId)
          nObject.put("swbfirstName",swbfirstName)
          nObject.put("taxFreeZone",taxFreeZone)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("title",title)
          nObject.put("unite_brand",unite_brand)
          nObject.put("userId",userId)

          nObject.put("address",address)
          nObject.put("administrative_region",administrative_region)
          nObject.put("aedzId",aedzId)
          nObject.put("city",city)
          nObject.put("city_grade",city_grade)
          nObject.put("city_origin",city_origin)
          nObject.put("district",district)
          nObject.put("district_origin",district_origin)
          nObject.put("economic_division",economic_division)
          nObject.put("if_city",if_city)
          nObject.put("if_district",if_district)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("latitude",latitude)
          nObject.put("longitude",longitude)
          nObject.put("name",name)
          nObject.put("poor_counties",poor_counties)
          nObject.put("province",province)
          nObject.put("regional_ID",regional_ID)
          nObject.put("registration_institution",registration_institution)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("shopId",shopId)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("town",town)
          nObject.put("urban_agglomerations",urban_agglomerations)

          nObject.toJSONString
        })

        spark.read.json(value).registerTempTable("frame")

        val shop = spark.sql(
          """
            |select
            |
            |timeStamp,
            |address,
            |administrative_region,
            |aedzId,
            |city,
            |city_grade,
            |city_origin,
            |district,
            |district_origin,
            |economic_division,
            |if_city,
            |if_district,
            |if_state_level_new_areas,
            |latitude,
            |longitude,
            |name,
            |poor_counties,
            |province,
            |regional_ID,
            |registration_institution,
            |rural_demonstration_counties,
            |rural_ecommerce,
            |shopId,
            |the_belt_and_road_city,
            |the_belt_and_road_province,
            |the_yangtze_river_economic_zone_city,
            |the_yangtze_river_economic_zone_province,
            |town,
            |urban_agglomerations
            |
            |from
            |frame
        """.stripMargin).dropDuplicates("shopId")
        shop.repartition(10).write.orc(s"s3a://dws-data/split/split_addressTable/${year}/${platform}/${month}")

        val data = spark.sql(
          """
            |select
            |
            |BC_type,
            |Base_Info,
            |allItemCount,
            |brandName,
            |brandName_cn,
            |brandName_en,
            |brandSourceId,
            |brandValueId,
            |brand_isLaoZiHao,
            |brand_state,
            |brand_type,
            |categoryId,
            |commentCount,
            |creditLevel,
            |delivery_from,
            |detailCategoryName,
            |evaluates,
            |fans,
            |favcount,
            |firstCategoryId,
            |fourthCategoryId,
            |goodRatePercentage,
            |goodUrl,
            |good_id,
            |images,
            |is_flagship,
            |is_market,
            |is_o2o,
            |is_onsell,
            |is_oversea,
            |originCountryName,
            |originItem,
            |originProduction,
            |original_cost,
            |platformId,
            |platformName,
            |priceText,
            |products,
            |quantity,
            |rootCategoryId,
            |salesAmount,
            |secondCategoryId,
            |sellCount,
            |sellerNick,
            |sellerType,
            |shopId,
            |shopName,
            |shopType,
            |shopUrl,
            |starts,
            |swb_products,
            |swbfirstId,
            |swbfirstName,
            |taxFreeZone,
            |thirdCategoryId,
            |timeStamp,
            |title,
            |unite_brand,
            |userId
            |
            |from
            |frame
        """.stripMargin).dropDuplicates("good_id")
        data.repartition(10).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

      }
    }
  }
  //京东
  def ES_ORC_JD(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {
      val frame = spark.read.orc(s"s3a://dws-data/g_data/${year}/${month}/jd/*")
      if(frame != null){
        val value = frame.toJSON.rdd.map(line => {
          val nObject = JSON.parseObject(line)
          val allCategoryId = nObject.getOrDefault("allCategoryId","-1").toString
          val allItemCount = nObject.getOrDefault("allItemCount","-1").toString
          val brandName = nObject.getOrDefault("brandName","-1").toString
          val brandName_cn = nObject.getOrDefault("brandName_cn","-1").toString
          val brandName_en = nObject.getOrDefault("brandName_en","-1").toString
          val brandSourceId = nObject.getOrDefault("brandSourceId","-1").toString
          val brandValueId = nObject.getOrDefault("brandValueId","-1").toString
          val brand_isLaoZiHao = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val brand_state = nObject.getOrDefault("brand_state","-1").toString
          val brand_type = nObject.getOrDefault("brand_type","-1").toString
          val categoryId = nObject.getOrDefault("categoryId","-1").toString
          val categoryName = nObject.getOrDefault("categoryName","-1").toString
          val commentCount = nObject.getOrDefault("commentCount","-1").toString
          val crawl_date = nObject.getOrDefault("crawl_date","-1").toString
          val delivery_from = nObject.getOrDefault("delivery_from","-1").toString
          val detailCategoryName = nObject.getOrDefault("detailCategoryName","-1").toString
          val evaluates = nObject.getOrDefault("evaluates","-1").toString
          val fans = nObject.getOrDefault("fans","-1").toString
          val firstCategoryId = nObject.getOrDefault("firstCategoryId","-1").toString
          val fourthCategoryId = nObject.getOrDefault("fourthCategoryId","-1").toString
          val generalCount = nObject.getOrDefault("generalCount","-1").toString
          val goodCount = nObject.getOrDefault("goodCount","-1").toString
          val goodRatePercentage = nObject.getOrDefault("goodRatePercentage","-1").toString
          val goodUrl = nObject.getOrDefault("goodUrl","-1").toString
          val good_id = nObject.getOrDefault("good_id","-1").toString
          val images = nObject.getOrDefault("images","-1").toString
          val is_ebook = nObject.getOrDefault("is_ebook","-1").toString
          val is_jingxuan = nObject.getOrDefault("is_jingxuan","-1").toString
          val is_market = nObject.getOrDefault("is_market","-1").toString
          val is_oversea = nObject.getOrDefault("is_oversea","-1").toString
          val is_pingou = nObject.getOrDefault("is_pingou","-1").toString
          val is_realestate = nObject.getOrDefault("is_realestate","-1").toString
          val is_selfSupport = nObject.getOrDefault("is_selfSupport","-1").toString
          val is_toplife = nObject.getOrDefault("is_toplife","-1").toString
          val originCountryName = nObject.getOrDefault("originCountryName","-1").toString
          val originItem = nObject.getOrDefault("originItem","-1").toString
          val originProduction = nObject.getOrDefault("originProduction","-1").toString
          val original_cost = nObject.getOrDefault("original_cost","-1").toString
          val platformId = nObject.getOrDefault("platformId","-1").toString
          val platformName = nObject.getOrDefault("platformName","-1").toString
          val poorCount = nObject.getOrDefault("poorCount","-1").toString
          val priceText = nObject.getOrDefault("priceText","-1").toString
          val products = nObject.getOrDefault("products","-1").toString
          val rootCategoryId = nObject.getOrDefault("rootCategoryId","-1").toString
          val rootCategoryName = nObject.getOrDefault("rootCategoryName","-1").toString
          val salesAmount = nObject.getOrDefault("salesAmount","-1").toString
          val score = nObject.getOrDefault("score","-1").toString
          val secondCategoryId = nObject.getOrDefault("secondCategoryId","-1").toString
          val sellCount = nObject.getOrDefault("sellCount","-1").toString
          val shopId = nObject.getOrDefault("shopId","-1").toString
          val shopName = nObject.getOrDefault("shopName","-1").toString
          val shopType = nObject.getOrDefault("shopType","-1").toString
          val shopUrl = nObject.getOrDefault("shopUrl","-1").toString
          val starts = nObject.getOrDefault("starts","-1").toString
          val subCategoryId = nObject.getOrDefault("subCategoryId","-1").toString
          val subCategoryName = nObject.getOrDefault("subCategoryName","-1").toString
          val swb_firstcategoryId = nObject.getOrDefault("swb_firstcategoryId","-1").toString
          val swb_products = nObject.getOrDefault("swb_products","-1").toString
          val swbfirstId = nObject.getOrDefault("swbfirstId","-1").toString
          val swbfirstName = nObject.getOrDefault("swbfirstName","-1").toString
          val taxFreeZone = nObject.getOrDefault("taxFreeZone","-1").toString
          val thirdCategoryId = nObject.getOrDefault("thirdCategoryId","-1").toString
          val timeStamp = nObject.getOrDefault("timeStamp","-1").toString
          val title = nObject.getOrDefault("title","-1").toString

          val address = nObject.getOrDefault("address","-1").toString
          val administrative_region = nObject.getOrDefault("administrative_region","-1").toString
          val aedzId = nObject.getOrDefault("aedzId","-1").toString
          val city = nObject.getOrDefault("city","-1").toString
          val city_grade = nObject.getOrDefault("city_grade","-1").toString
          val city_origin = nObject.getOrDefault("city_origin","-1").toString
          val district = nObject.getOrDefault("district","-1").toString
          val district_origin = nObject.getOrDefault("district_origin","-1").toString
          val economic_division = nObject.getOrDefault("economic_division","-1").toString
          val if_city = nObject.getOrDefault("if_city","-1").toString
          val if_district = nObject.getOrDefault("if_district","-1").toString
          val if_state_level_new_areas = nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val latitude = nObject.getOrDefault("latitude","-1").toString
          val longitude = nObject.getOrDefault("longitude","-1").toString
          val name = nObject.getOrDefault("name","-1").toString
          val poor_counties = nObject.getOrDefault("poor_counties","-1").toString
          val province = nObject.getOrDefault("province","-1").toString
          val regional_ID = nObject.getOrDefault("regional_ID","-1").toString
          val registration_institution = nObject.getOrDefault("registration_institution","-1").toString
          val rural_demonstration_counties = nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val rural_ecommerce = nObject.getOrDefault("rural_ecommerce","-1").toString
          val the_belt_and_road_city = nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val the_belt_and_road_province = nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val the_yangtze_river_economic_zone_city = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val the_yangtze_river_economic_zone_province = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val town = nObject.getOrDefault("town","-1").toString
          val urban_agglomerations = nObject.getOrDefault("urban_agglomerations","-1").toString

          //41
          nObject.put("allCategoryId",allCategoryId)
          nObject.put("allItemCount",allItemCount)
          nObject.put("brandName",brandName)
          nObject.put("brandName_cn",brandName_cn)
          nObject.put("brandName_en",brandName_en)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("brandValueId",brandValueId)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("brand_state",brand_state)
          nObject.put("brand_type",brand_type)
          nObject.put("categoryId",categoryId)
          nObject.put("categoryName",categoryName)
          nObject.put("commentCount",commentCount)
          nObject.put("crawl_date",crawl_date)
          nObject.put("delivery_from",delivery_from)
          nObject.put("detailCategoryName",detailCategoryName)
          nObject.put("evaluates",evaluates)
          nObject.put("fans",fans)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("generalCount",generalCount)
          nObject.put("goodCount",goodCount)
          nObject.put("goodRatePercentage",goodRatePercentage)
          nObject.put("goodUrl",goodUrl)
          nObject.put("good_id",good_id)
          nObject.put("images",images)
          nObject.put("is_ebook",is_ebook)
          nObject.put("is_jingxuan",is_jingxuan)
          nObject.put("is_market",is_market)
          nObject.put("is_oversea",is_oversea)
          nObject.put("is_pingou",is_pingou)
          nObject.put("is_realestate",is_realestate)
          nObject.put("is_selfSupport",is_selfSupport)
          nObject.put("is_toplife",is_toplife)
          nObject.put("originCountryName",originCountryName)
          nObject.put("originItem",originItem)
          nObject.put("originProduction",originProduction)
          nObject.put("original_cost",original_cost)
          nObject.put("platformId",platformId)
          nObject.put("platformName",platformName)
          nObject.put("poorCount",poorCount)
          nObject.put("priceText",priceText)
          nObject.put("products",products)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("salesAmount",salesAmount)
          nObject.put("score",score)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("sellCount",sellCount)
          nObject.put("shopId",shopId)
          nObject.put("shopName",shopName)
          nObject.put("shopType",shopType)
          nObject.put("shopUrl",shopUrl)
          nObject.put("starts",starts)
          nObject.put("subCategoryId",subCategoryId)
          nObject.put("subCategoryName",subCategoryName)
          nObject.put("swb_firstcategoryId",swb_firstcategoryId)
          nObject.put("swb_products",swb_products)
          nObject.put("swbfirstId",swbfirstId)
          nObject.put("swbfirstName",swbfirstName)
          nObject.put("taxFreeZone",taxFreeZone)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("title",title)



          nObject.put("address",address)
          nObject.put("administrative_region",administrative_region)
          nObject.put("aedzId",aedzId)
          nObject.put("city",city)
          nObject.put("city_grade",city_grade)
          nObject.put("city_origin",city_origin)
          nObject.put("district",district)
          nObject.put("district_origin",district_origin)
          nObject.put("economic_division",economic_division)
          nObject.put("if_city",if_city)
          nObject.put("if_district",if_district)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("latitude",latitude)
          nObject.put("longitude",longitude)
          nObject.put("name",name)
          nObject.put("poor_counties",poor_counties)
          nObject.put("province",province)
          nObject.put("regional_ID",regional_ID)
          nObject.put("registration_institution",registration_institution)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("town",town)
          nObject.put("urban_agglomerations",urban_agglomerations)


          nObject.toJSONString
        })

        spark.read.json(value).registerTempTable("frame")
        val data = spark.sql(
          """
            |select
            |
            |allCategoryId,
            |allItemCount,
            |brandName,
            |brandName_cn,
            |brandName_en,
            |brandSourceId,
            |brandValueId,
            |brand_isLaoZiHao,
            |brand_state,
            |brand_type,
            |categoryId,
            |categoryName,
            |commentCount,
            |crawl_date,
            |delivery_from,
            |detailCategoryName,
            |evaluates,
            |fans,
            |firstCategoryId,
            |fourthCategoryId,
            |generalCount,
            |goodCount,
            |goodRatePercentage,
            |goodUrl,
            |good_id,
            |images,
            |is_ebook,
            |is_jingxuan,
            |is_market,
            |is_oversea,
            |is_pingou,
            |is_realestate,
            |is_selfSupport,
            |is_toplife,
            |originCountryName,
            |originItem,
            |originProduction,
            |original_cost,
            |platformId,
            |platformName,
            |poorCount,
            |priceText,
            |products,
            |rootCategoryId,
            |rootCategoryName,
            |salesAmount,
            |score,
            |secondCategoryId,
            |sellCount,
            |shopId,
            |shopName,
            |shopType,
            |shopUrl,
            |starts,
            |subCategoryId,
            |subCategoryName,
            |swb_firstcategoryId,
            |swb_products,
            |swbfirstId,
            |swbfirstName,
            |taxFreeZone,
            |thirdCategoryId,
            |timeStamp,
            |title
            |
            |from
            |frame
        """.stripMargin).dropDuplicates("good_id")
        data.repartition(10).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")
        val shop = spark.sql(
          """
            |select
            |
            |timeStamp,
            |address,
            |administrative_region,
            |aedzId,
            |city,
            |city_grade,
            |city_origin,
            |district,
            |district_origin,
            |economic_division,
            |if_city,
            |if_district,
            |if_state_level_new_areas,
            |latitude,
            |longitude,
            |name,
            |poor_counties,
            |province,
            |regional_ID,
            |registration_institution,
            |rural_demonstration_counties,
            |rural_ecommerce,
            |shopId,
            |the_belt_and_road_city,
            |the_belt_and_road_province,
            |the_yangtze_river_economic_zone_city,
            |the_yangtze_river_economic_zone_province,
            |town,
            |urban_agglomerations
            |
            |from
            |frame
        """.stripMargin).dropDuplicates("shopId")

        shop.repartition(32).write.orc(s"s3a://dws-data/split/split_addressTable/${year}/${platform}/${month}")

      }
    }
  }
  //美团
  def ES_ORC_MT(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {
      val frame = spark.read.orc(s"s3a://dws-data/g_data/${year}/${month}/meituan/*")
      if(frame != null){
        val value = frame.toJSON.rdd.map(line => {
          val nObject = JSON.parseObject(line)
          val obj = new JSONObject()

          val categoryId   = nObject.getOrDefault("categoryId","-1").toString
          val categoryName   = nObject.getOrDefault("categoryName","-1").toString
          val county   = nObject.getOrDefault("county","-1").toString
          val delivery_fee   = nObject.getOrDefault("delivery_fee","-1").toString
          val delivery_mode   = nObject.getOrDefault("delivery_mode","-1").toString
          val dpShopId   = nObject.getOrDefault("dpShopId","-1").toString
          val emotionalKeywords   = nObject.getOrDefault("emotionalKeywords","-1").toString
          val evaluates   = nObject.getOrDefault("evaluates","-1").toString
          val firstCategoryId   = nObject.getOrDefault("firstCategoryId","-1").toString
          val flavors   = nObject.getOrDefault("flavors","-1").toString
          val food_type   = nObject.getOrDefault("food_type","-1").toString
          val fourthCategoryId   = nObject.getOrDefault("fourthCategoryId","-1").toString
          val goodDescription   = nObject.getOrDefault("goodDescription","-1").toString
          val goodRatePercentage   = nObject.getOrDefault("goodRatePercentage","-1").toString
          val good_id   = nObject.getOrDefault("good_id","-1").toString
          val images   = nObject.getOrDefault("images","-1").toString
          val is_premium   = nObject.getOrDefault("is_premium","-1").toString
          val licencePics   = nObject.getOrDefault("licencePics","-1").toString
          val min_delivery_price   = nObject.getOrDefault("min_delivery_price","-1").toString
          val mtWmPoiId   = nObject.getOrDefault("mtWmPoiId","-1").toString
          val opening_hours   = nObject.getOrDefault("opening_hours","-1").toString
          val order_lead_time   = nObject.getOrDefault("order_lead_time","-1").toString
          val original_cost   = nObject.getOrDefault("original_cost","-1").toString
          val packing_fee   = nObject.getOrDefault("packing_fee","-1").toString
          val phone   = nObject.getOrDefault("phone","-1").toString
          val platformId   = nObject.getOrDefault("platformId","-1").toString
          val platformName   = nObject.getOrDefault("platformName","-1").toString
          val praiseNum   = nObject.getOrDefault("praiseNum","-1").toString
          val priceText   = nObject.getOrDefault("priceText","-1").toString.toDouble
          val promotion_info   = nObject.getOrDefault("promotion_info","-1").toString
          val rootCategoryId   = nObject.getOrDefault("rootCategoryId","-1").toString
          val rootCategoryName   = nObject.getOrDefault("rootCategoryName","-1").toString
          val salesAmount   = nObject.getOrDefault("salesAmount","-1").toString.toDouble
          val secondCategoryId   = nObject.getOrDefault("secondCategoryId","-1").toString
          val sellCount   = nObject.getOrDefault("sellCount","-1").toString.toDouble.toLong
          val shopCommentCount   = nObject.getOrDefault("shopCommentCount","-1").toString
          val shopDescription   = nObject.getOrDefault("shopDescription","-1").toString
          val shopId   = nObject.getOrDefault("shopId","-1").toString
          val shopImages   = nObject.getOrDefault("shopImages","-1").toString
          val shopName   = nObject.getOrDefault("shopName","-1").toString
          val shopSellCount   = nObject.getOrDefault("shopSellCount","-1").toString
          val shopUrl   = nObject.getOrDefault("shopUrl","-1").toString
          val street   = nObject.getOrDefault("street","-1").toString
          val thirdCategoryId   = nObject.getOrDefault("thirdCategoryId","-1").toString
          val third_category_name   = nObject.getOrDefault("third_category_name","-1").toString
          val timeStamp   = nObject.getOrDefault("timeStamp","-1").toString
          val title   = nObject.getOrDefault("title","-1").toString
          val town = nObject.getOrDefault("town", "-1").toString
          val aedzId = nObject.getOrDefault("aedzId", "-1").toString
          val address = nObject.getOrDefault("address", "-1").toString
          val regional_ID = nObject.getOrDefault("regional_ID", "-1").toString
          val administrative_region = nObject.getOrDefault("administrative_region", "-1").toString


          obj.put("town", town)
          obj.put("administrative_region", administrative_region)
          obj.put("regional_ID", regional_ID)
          obj.put("address", address)
          obj.put("aedzId", aedzId)
          obj.put("categoryId",categoryId)
          obj.put("categoryName",categoryName)
          obj.put("county",county)
          obj.put("delivery_fee",delivery_fee)
          obj.put("delivery_mode",delivery_mode)
          obj.put("dpShopId",dpShopId)
          obj.put("emotionalKeywords",emotionalKeywords)
          obj.put("evaluates",evaluates)
          obj.put("firstCategoryId",firstCategoryId)
          obj.put("flavors",flavors)
          obj.put("food_type",food_type)
          obj.put("fourthCategoryId",fourthCategoryId)
          obj.put("goodDescription",goodDescription)
          obj.put("goodRatePercentage",goodRatePercentage)
          obj.put("good_id",good_id)
          obj.put("images",images)
          obj.put("is_premium",is_premium)
          obj.put("licencePics",licencePics)
          obj.put("min_delivery_price",min_delivery_price)
          obj.put("mtWmPoiId",mtWmPoiId)
          obj.put("opening_hours",opening_hours)
          obj.put("order_lead_time",order_lead_time)
          obj.put("original_cost",original_cost)
          obj.put("packing_fee",packing_fee)
          obj.put("phone",phone)
          obj.put("platformId",platformId)
          obj.put("platformName",platformName)
          obj.put("praiseNum",praiseNum)
          obj.put("priceText",priceText)
          obj.put("promotion_info",promotion_info)
          obj.put("rootCategoryId",rootCategoryId)
          obj.put("rootCategoryName",rootCategoryName)
          obj.put("salesAmount",salesAmount)
          obj.put("secondCategoryId",secondCategoryId)
          obj.put("sellCount",sellCount)
          obj.put("shopCommentCount",shopCommentCount)
          obj.put("shopDescription",shopDescription)
          obj.put("shopId",shopId)
          obj.put("shopImages",shopImages)
          obj.put("shopName",shopName)
          obj.put("shopSellCount",shopSellCount)
          obj.put("shopUrl",shopUrl)
          obj.put("street",street)
          obj.put("thirdCategoryId",thirdCategoryId)
          obj.put("third_category_name",third_category_name)
          obj.put("timeStamp",timeStamp)
          obj.put("title",title)


          obj.toJSONString
        })
        spark.read.json(value).registerTempTable("frame")

        val shop = spark.sql(
          """
            |select
            |
            |timeStamp,
            |address,
            |administrative_region,
            |aedzId,
            |city,
            |city_grade,
            |city_origin,
            |district,
            |district_origin,
            |economic_division,
            |if_city,
            |if_district,
            |if_state_level_new_areas,
            |latitude,
            |longitude,
            |poor_counties,
            |province,
            |regional_ID,
            |rural_demonstration_counties,
            |rural_ecommerce,
            |shopId,
            |the_belt_and_road_city,
            |the_belt_and_road_province,
            |the_yangtze_river_economic_zone_city,
            |the_yangtze_river_economic_zone_province,
            |town,
            |urban_agglomerations
            |
            |from
            |frame
        """.stripMargin).dropDuplicates("shopId")
        shop.repartition(10).write.orc(s"s3a://dws-data/split/split_addressTable/${year}/${platform}/${month}")

        val data = spark.sql(
          """
            |select
            |
            |categoryId,
            |categoryName,
            |county,
            |delivery_fee,
            |delivery_mode,
            |dpShopId,
            |emotionalKeywords,
            |evaluates,
            |firstCategoryId,
            |flavors,
            |food_type,
            |fourthCategoryId,
            |goodDescription,
            |goodRatePercentage,
            |good_id,
            |images,
            |is_premium,
            |licencePics,
            |min_delivery_price,
            |mtWmPoiId,
            |opening_hours,
            |order_lead_time,
            |original_cost,
            |packing_fee,
            |phone,
            |platformId,
            |platformName,
            |praiseNum,
            |priceText,
            |promotion_info,
            |rootCategoryId,
            |rootCategoryName,
            |salesAmount,
            |secondCategoryId,
            |sellCount,
            |shopCommentCount,
            |shopDescription,
            |shopId,
            |shopImages,
            |shopName,
            |shopSellCount,
            |shopUrl,
            |street,
            |thirdCategoryId,
            |third_category_name,
            |timeStamp,
            |title
            |
            |from
            |frame
        """.stripMargin)//.dropDuplicates("good_id")
        data.repartition(10).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

      }
    }
  }
  //饿了么
  def ES_ORC_ELM(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {
      val frame = spark.read.orc(s"s3a://dws-data/g_data/${year}/${month}/${platform}/*")
      if(frame != null){
        val value = frame.toJSON.rdd.map(line => {
          val nObject = JSON.parseObject(line)

          val allItemCount = nObject.getOrDefault("allItemCount", "-1").toString
          val commentCount = nObject.getOrDefault("commentCount", "-1").toString
          val county = nObject.getOrDefault("county", "-1").toString
          val crawl_date = nObject.getOrDefault("crawl_date", "-1").toString
          val delivery_fee = nObject.getOrDefault("delivery_fee", "-1").toString
          val delivery_mode = nObject.getOrDefault("delivery_mode", "-1").toString
          val dpShopId = nObject.getOrDefault("dpShopId", "-1").toString
          val emotionalKeywords = nObject.getOrDefault("emotionalKeywords", "-1").toString
          val evaluates = nObject.getOrDefault("evaluates", "-1").toString
          val firstCategoryId = nObject.getOrDefault("firstCategoryId", "-1").toString
          val flavors = nObject.getOrDefault("flavors", "-1").toString
          val food_type = nObject.getOrDefault("food_type", "-1").toString
          val fourthCategoryId = nObject.getOrDefault("fourthCategoryId", "-1").toString
          val goodRatePercentage = nObject.getOrDefault("goodRatePercentage", "-1").toString
          val good_id = nObject.getOrDefault("good_id", "-1").toString
          val is_premium = nObject.getOrDefault("is_premium", "-1").toString
          val min_delivery_price = nObject.getOrDefault("min_delivery_price", "-1").toString
          val opening_hours = nObject.getOrDefault("opening_hours", "-1").toString
          val order_lead_time = nObject.getOrDefault("order_lead_time", "-1").toString
          val original_cost = nObject.getOrDefault("original_cost", "-1").toString
          val packing_fee = nObject.getOrDefault("packing_fee", "-1").toString
          val phone = nObject.getOrDefault("phone", "-1").toString
          val platformId = nObject.getOrDefault("platformId", "-1").toString
          val platformName = nObject.getOrDefault("platformName", "-1").toString
          val platformName_spelling = nObject.getOrDefault("platformName_spelling", "-1").toString
          val priceText = nObject.getOrDefault("priceText", "-1").toString
          val rating = nObject.getOrDefault("rating", "-1").toString
          val rootCategoryId = nObject.getOrDefault("rootCategoryId", "-1").toString
          val rootCategoryName = nObject.getOrDefault("rootCategoryName", "-1").toString
          val salesAmount = nObject.getOrDefault("salesAmount", "-1").toString
          val secondCategoryId = nObject.getOrDefault("secondCategoryId", "-1").toString
          val sellCount = nObject.getOrDefault("sellCount", "-1").toString
          val shopCommentCount = nObject.getOrDefault("shopCommentCount", "-1").toString
          val shopDescription = nObject.getOrDefault("shopDescription", "-1").toString
          val shopId = nObject.getOrDefault("shopId", "-1").toString
          val shopImages = nObject.getOrDefault("shopImages", "-1").toString
          val shopName = nObject.getOrDefault("shopName", "-1").toString
          val shopSellCount = nObject.getOrDefault("shopSellCount", "-1").toString
          val shopType = nObject.getOrDefault("shopType", "-1").toString
          val shopUrl = nObject.getOrDefault("shopUrl", "-1").toString
          val street = nObject.getOrDefault("street", "-1").toString
          val thirdCategoryId = nObject.getOrDefault("thirdCategoryId", "-1").toString
          val timeStamp = nObject.getOrDefault("timeStamp", "-1").toString
          val title = nObject.getOrDefault("title", "-1").toString
          val town = nObject.getOrDefault("town", "-1").toString
          val aedzId = nObject.getOrDefault("aedzId", "-1").toString
          val address = nObject.getOrDefault("address", "-1").toString

          nObject.put("allItemCount", allItemCount)
          nObject.put("address", address)
          nObject.put("town", town)
          nObject.put("aedzId", aedzId)
          nObject.put("commentCount", commentCount)
          nObject.put("county", county)
          nObject.put("crawl_date", crawl_date)
          nObject.put("delivery_fee", delivery_fee)
          nObject.put("delivery_mode", delivery_mode)
          nObject.put("dpShopId", dpShopId)
          nObject.put("emotionalKeywords", emotionalKeywords)
          nObject.put("evaluates", evaluates)
          nObject.put("firstCategoryId", firstCategoryId)
          nObject.put("flavors", flavors)
          nObject.put("food_type", food_type)
          nObject.put("fourthCategoryId", fourthCategoryId)
          nObject.put("goodRatePercentage", goodRatePercentage)
          nObject.put("good_id", good_id)
          nObject.put("is_premium", is_premium)
          nObject.put("min_delivery_price", min_delivery_price)
          nObject.put("opening_hours", opening_hours)
          nObject.put("order_lead_time", order_lead_time)
          nObject.put("original_cost", original_cost)
          nObject.put("packing_fee", packing_fee)
          nObject.put("phone", phone)
          nObject.put("platformId", platformId)
          nObject.put("platformName", platformName)
          nObject.put("platformName_spelling", platformName_spelling)
          nObject.put("priceText", priceText)
          nObject.put("rating", rating)
          nObject.put("rootCategoryId", rootCategoryId)
          nObject.put("rootCategoryName", rootCategoryName)
          nObject.put("salesAmount", salesAmount)
          nObject.put("secondCategoryId", secondCategoryId)
          nObject.put("sellCount", sellCount)
          nObject.put("shopCommentCount", shopCommentCount)
          nObject.put("shopDescription", shopDescription)
          nObject.put("shopId", shopId)
          nObject.put("shopImages", shopImages)
          nObject.put("shopName", shopName)
          nObject.put("shopSellCount", shopSellCount)
          nObject.put("shopType", shopType)
          nObject.put("shopUrl", shopUrl)
          nObject.put("street", street)
          nObject.put("thirdCategoryId", thirdCategoryId)
          nObject.put("timeStamp", timeStamp)
          nObject.put("title", title)


          nObject.toJSONString
        })
        spark.read.json(value).registerTempTable("frame")
        val shop = spark.sql(
          """
            |select
            |
            |timeStamp,
            |address,
            |administrative_region,
            |aedzId,
            |city,
            |city_grade,
            |city_origin,
            |district,
            |district_origin,
            |economic_division,
            |if_city,
            |if_district,
            |if_state_level_new_areas,
            |latitude,
            |longitude,
            |poor_counties,
            |province,
            |regional_ID,
            |rural_demonstration_counties,
            |rural_ecommerce,
            |shopId,
            |the_belt_and_road_city,
            |the_belt_and_road_province,
            |the_yangtze_river_economic_zone_city,
            |the_yangtze_river_economic_zone_province,
            |town,
            |urban_agglomerations
            |
            |from
            |frame
        """.stripMargin).dropDuplicates("shopId")
        shop.repartition(10).write.orc(s"s3a://dws-data/split/split_addressTable/${year}/${platform}/${month}")
        val data = spark.sql(
          """
            |select
            |
            |allItemCount,
            |commentCount,
            |county,
            |crawl_date,
            |delivery_fee,
            |delivery_mode,
            |dpShopId,
            |emotionalKeywords,
            |evaluates,
            |firstCategoryId,
            |flavors,
            |food_type,
            |fourthCategoryId,
            |goodRatePercentage,
            |good_id,
            |is_premium,
            |min_delivery_price,
            |opening_hours,
            |order_lead_time,
            |original_cost,
            |packing_fee,
            |phone,
            |platformId,
            |platformName,
            |platformName_spelling,
            |priceText,
            |rating,
            |rootCategoryId,
            |rootCategoryName,
            |salesAmount,
            |secondCategoryId,
            |sellCount,
            |shopCommentCount,
            |shopDescription,
            |shopId,
            |shopImages,
            |shopName,
            |shopSellCount,
            |shopType,
            |shopUrl,
            |street,
            |thirdCategoryId,
            |timeStamp,
            |title
            |
            |from
            |frame
        """.stripMargin).dropDuplicates("good_id")
        data.repartition(10).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

      }
    }
  }
  //苏宁
  def ES_ORC_SUNING(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
  val sc = spark.sparkContext
  for (month <- monthArr) {
    val frame = spark.read.orc(s"s3a://dws-data/g_data/${year}/${month}/suning/*")
    if(frame != null){
      val value = frame.toJSON.rdd.map(line => {
        val nObject = JSON.parseObject(line)
        val Base_Info = nObject.getOrDefault("Base_Info","-1").toString
        val brandName = nObject.getOrDefault("brandName","-1").toString
        val brandName_cn = nObject.getOrDefault("brandName_cn","-1").toString
        val brandName_en = nObject.getOrDefault("brandName_en","-1").toString
        val brandSourceId = nObject.getOrDefault("brandSourceId","-1").toString
        val brandValueId = nObject.getOrDefault("brandValueId","-1").toString
        val brand_isLaoZiHao = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
        val brand_state = nObject.getOrDefault("brand_state","-1").toString
        val brand_type = nObject.getOrDefault("brand_type","-1").toString
        val categoryId = nObject.getOrDefault("categoryId","-1").toString
        val categoryName = nObject.getOrDefault("categoryName","-1").toString
        val color = nObject.getOrDefault("color","-1").toString
        val commentCount = nObject.getOrDefault("commentCount","-1").toString
        val crawl_date = nObject.getOrDefault("crawl_date","-1").toString
        val delivery_from = nObject.getOrDefault("delivery_from","-1").toString
        val detailCategoryName = nObject.getOrDefault("detailCategoryName","-1").toString
        val earnest = nObject.getOrDefault("earnest","-1").toString
        val evaluates = nObject.getOrDefault("evaluates","-1").toString
        val finalPayment = nObject.getOrDefault("finalPayment","-1").toString
        val firstCategoryId = nObject.getOrDefault("firstCategoryId","-1").toString
        val fourthCategoryId = nObject.getOrDefault("fourthCategoryId","-1").toString
        val generalCount = nObject.getOrDefault("generalCount","-1").toString
        val goodCount = nObject.getOrDefault("goodCount","-1").toString
        val goodRatePercentage = nObject.getOrDefault("goodRatePercentage","-1").toString
        val goodUrl = nObject.getOrDefault("goodUrl","-1").toString
        val good_id = nObject.getOrDefault("good_id","-1").toString
        val is_market = nObject.getOrDefault("is_market","-1").toString
        val is_o2o = nObject.getOrDefault("is_o2o","-1").toString
        val is_oversea = nObject.getOrDefault("is_oversea","-1").toString
        val is_selfSupport = nObject.getOrDefault("is_selfSupport","-1").toString
        val model = nObject.getOrDefault("model","-1").toString
        val originCountryName = nObject.getOrDefault("originCountryName","-1").toString
        val originItem = nObject.getOrDefault("originItem","-1").toString
        val originProduction = nObject.getOrDefault("originProduction","-1").toString
        val original_cost = nObject.getOrDefault("original_cost","-1").toString
        val platformId = nObject.getOrDefault("platformId","-1").toString
        val platformName = nObject.getOrDefault("platformName","-1").toString
        val poorCount = nObject.getOrDefault("poorCount","-1").toString
        val preSaleCount = nObject.getOrDefault("preSaleCount","-1").toString
        val priceText = nObject.getOrDefault("priceText","-1").toString
        val products = nObject.getOrDefault("products","-1").toString
        val ram = nObject.getOrDefault("ram","-1").toString
        val rom = nObject.getOrDefault("rom","-1").toString
        val rootCategoryId = nObject.getOrDefault("rootCategoryId","-1").toString
        val rootCategoryName = nObject.getOrDefault("rootCategoryName","-1").toString
        val salesAmount = nObject.getOrDefault("salesAmount","-1").toString
        val score = nObject.getOrDefault("score","-1").toString
        val secondCategoryId = nObject.getOrDefault("secondCategoryId","-1").toString
        val sellCount = nObject.getOrDefault("sellCount","-1").toString
        val series = nObject.getOrDefault("series","-1").toString
        val shopId = nObject.getOrDefault("shopId","-1").toString
        val shopName = nObject.getOrDefault("shopName","-1").toString
        val shopType = nObject.getOrDefault("shopType","-1").toString
        val shopUrl = nObject.getOrDefault("shopUrl","-1").toString
        val shop_good_id = nObject.getOrDefault("shop_good_id","-1").toString
        val subCategoryId = nObject.getOrDefault("subCategoryId","-1").toString
        val subCategoryName = nObject.getOrDefault("subCategoryName","-1").toString
        val taxFreeZone = nObject.getOrDefault("taxFreeZone","-1").toString
        val thirdCategoryId = nObject.getOrDefault("thirdCategoryId","-1").toString
        val timeStamp = nObject.getOrDefault("timeStamp","-1").toString
        val title = nObject.getOrDefault("title","-1").toString
        val town = nObject.getOrDefault("town", "-1").toString
        val aedzId = nObject.getOrDefault("aedzId", "-1").toString
        val address = nObject.getOrDefault("address", "-1").toString
        val name = nObject.getOrDefault("name", "-1").toString

        nObject.put("town", town)
        nObject.put("address", address)
        nObject.put("name", name)
        nObject.put("aedzId", aedzId)
        nObject.put("Base_Info",Base_Info)
        nObject.put("brandName",brandName)
        nObject.put("brandName_cn",brandName_cn)
        nObject.put("brandName_en",brandName_en)
        nObject.put("brandSourceId",brandSourceId)
        nObject.put("brandValueId",brandValueId)
        nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
        nObject.put("brand_state",brand_state)
        nObject.put("brand_type",brand_type)
        nObject.put("categoryId",categoryId)
        nObject.put("categoryName",categoryName)
        nObject.put("color",color)
        nObject.put("commentCount",commentCount)
        nObject.put("crawl_date",crawl_date)
        nObject.put("delivery_from",delivery_from)
        nObject.put("detailCategoryName",detailCategoryName)
        nObject.put("earnest",earnest)
        nObject.put("evaluates",evaluates)
        nObject.put("finalPayment",finalPayment)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("generalCount",generalCount)
        nObject.put("goodCount",goodCount)
        nObject.put("goodRatePercentage",goodRatePercentage)
        nObject.put("goodUrl",goodUrl)
        nObject.put("good_id",good_id)
        nObject.put("is_market",is_market)
        nObject.put("is_o2o",is_o2o)
        nObject.put("is_oversea",is_oversea)
        nObject.put("is_selfSupport",is_selfSupport)
        nObject.put("model",model)
        nObject.put("originCountryName",originCountryName)
        nObject.put("originItem",originItem)
        nObject.put("originProduction",originProduction)
        nObject.put("original_cost",original_cost)
        nObject.put("platformId",platformId)
        nObject.put("platformName",platformName)
        nObject.put("poorCount",poorCount)
        nObject.put("preSaleCount",preSaleCount)
        nObject.put("priceText",priceText)
        nObject.put("products",products)
        nObject.put("ram",ram)
        nObject.put("rom",rom)
        nObject.put("rootCategoryId",rootCategoryId)
        nObject.put("rootCategoryName",rootCategoryName)
        nObject.put("salesAmount",salesAmount)
        nObject.put("score",score)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("sellCount",sellCount)
        nObject.put("series",series)
        nObject.put("shopId",shopId)
        nObject.put("shopName",shopName)
        nObject.put("shopType",shopType)
        nObject.put("shopUrl",shopUrl)
        nObject.put("shop_good_id",shop_good_id)
        nObject.put("subCategoryId",subCategoryId)
        nObject.put("subCategoryName",subCategoryName)
        nObject.put("taxFreeZone",taxFreeZone)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("title",title)

        nObject.toJSONString
      })

      spark.read.json(value).registerTempTable("frame")

      val data = spark.sql(
        """
          |select
          |
          |Base_Info,
          |brandName,
          |brandName_cn,
          |brandName_en,
          |brandSourceId,
          |brandValueId,
          |brand_isLaoZiHao,
          |brand_state,
          |brand_type,
          |categoryId,
          |categoryName,
          |color,
          |commentCount,
          |crawl_date,
          |delivery_from,
          |detailCategoryName,
          |earnest,
          |evaluates,
          |finalPayment,
          |firstCategoryId,
          |fourthCategoryId,
          |generalCount,
          |goodCount,
          |goodRatePercentage,
          |goodUrl,
          |good_id,
          |is_market,
          |is_o2o,
          |is_oversea,
          |is_selfSupport,
          |model,
          |originCountryName,
          |originItem,
          |originProduction,
          |original_cost,
          |platformId,
          |platformName,
          |poorCount,
          |preSaleCount,
          |priceText,
          |products,
          |ram,
          |rom,
          |rootCategoryId,
          |rootCategoryName,
          |salesAmount,
          |score,
          |secondCategoryId,
          |sellCount,
          |series,
          |shopId,
          |shopName,
          |shopType,
          |shopUrl,
          |shop_good_id,
          |subCategoryId,
          |subCategoryName,
          |taxFreeZone,
          |thirdCategoryId,
          |timeStamp,
          |title
          |
          |from
          |frame
        """.stripMargin).dropDuplicates("good_id")
      data.repartition(10).write.orc(s"s3a://dws-data/split/split_data/${year}/suning/${month}")

      val shop = spark.sql(
        """
          |select
          |
          |timeStamp,
          |address,
          |administrative_region,
          |aedzId,
          |city,
          |city_grade,
          |city_origin,
          |economic_division,
          |if_city,
          |if_district,
          |if_state_level_new_areas,
          |latitude,
          |longitude,
          |name,
          |poor_counties,
          |province,
          |registration_institution,
          |rural_demonstration_counties,
          |rural_ecommerce,
          |shopId,
          |the_belt_and_road_city,
          |the_belt_and_road_province,
          |the_yangtze_river_economic_zone_city,
          |the_yangtze_river_economic_zone_province,
          |town,
          |urban_agglomerations,
          |regional_ID,
          |district,
          |district_origin
          |
          |from
          |frame
        """.stripMargin).dropDuplicates("shopId")

      shop.repartition(10).write.orc(s"s3a://dws-data/split/split_addressTable/${year}/suning/${month}")
    }
  }
}
  //拼多多
  def ES_ORC_PDD(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {
      val frame = spark.read.orc(s"s3a://dws-data/g_data/${year}/${month}/pinduoduo/*")
      if(frame != null){
        val value = frame.toJSON.rdd.map(line => {
            val nObject = JSON.parseObject(line)

            val allItemCount = nObject.getOrDefault("allItemCount","-1").toString
            val categoryId = nObject.getOrDefault("categoryId","-1").toString
            val categoryName = nObject.getOrDefault("categoryName","-1").toString
            val commentCount = nObject.getOrDefault("commentCount","-1").toString
            val crawl_date = nObject.getOrDefault("crawl_date","-1").toString
            val emotionalKeywords = nObject.getOrDefault("emotionalKeywords","-1").toString
            val evaluates = nObject.getOrDefault("evaluates","-1").toString
            val firstCategoryId = nObject.getOrDefault("firstCategoryId","-1").toString
            val fourthCategoryId = nObject.getOrDefault("fourthCategoryId","-1").toString
            val goodUrl = nObject.getOrDefault("goodUrl","-1").toString
            val good_id = nObject.getOrDefault("good_id","-1").toString
            val group_num = nObject.getOrDefault("group_num","-1").toString
            val group_price = nObject.getOrDefault("group_price","-1").toString
            val is_oversea = nObject.getOrDefault("is_oversea","-1").toString
            val name = nObject.getOrDefault("name","-1").toString
            val originProduction = nObject.getOrDefault("originProduction","-1").toString
            val original_cost = nObject.getOrDefault("original_cost","-1").toString
            val phone = nObject.getOrDefault("phone","-1").toString
            val platformId = nObject.getOrDefault("platformId","-1").toString
            val platformName = nObject.getOrDefault("platformName","-1").toString
            val priceText = nObject.getOrDefault("priceText","-1").toString
            val products = nObject.getOrDefault("products","-1").toString
            val region_id = nObject.getOrDefault("region_id","-1").toString
            val rootCategoryId = nObject.getOrDefault("rootCategoryId","-1").toString
            val rootCategoryName = nObject.getOrDefault("rootCategoryName","-1").toString
            val salesAmount = nObject.getOrDefault("salesAmount","-1").toString
            val secondCategoryId = nObject.getOrDefault("secondCategoryId","-1").toString
            val sellCount = nObject.getOrDefault("sellCount","-1").toString
            val serviceCommitment = nObject.getOrDefault("serviceCommitment","-1").toString
            val shopId = nObject.getOrDefault("shopId","-1").toString
            val shopName = nObject.getOrDefault("shopName","-1").toString
            val shopSales = nObject.getOrDefault("shopSales","-1").toString
            val shopType = nObject.getOrDefault("shopType","-1").toString
            val shopUrl = nObject.getOrDefault("shopUrl","-1").toString
            val subCategoryId = nObject.getOrDefault("subCategoryId","-1").toString
            val subCategoryName = nObject.getOrDefault("subCategoryName","-1").toString
            val thirdCategoryId = nObject.getOrDefault("thirdCategoryId","-1").toString
            val timeStamp = nObject.getOrDefault("timeStamp","-1").toString
            val title = nObject.getOrDefault("title","-1").toString
            val town = nObject.getOrDefault("town", "-1").toString
            val aedzId = nObject.getOrDefault("aedzId", "-1").toString


            nObject.put("town", town)
            nObject.put("aedzId", aedzId)
            nObject.put("allItemCount",allItemCount)
            nObject.put("categoryId",categoryId)
            nObject.put("categoryName",categoryName)
            nObject.put("commentCount",commentCount)
            nObject.put("crawl_date",crawl_date)
            nObject.put("emotionalKeywords",emotionalKeywords)
            nObject.put("evaluates",evaluates)
            nObject.put("firstCategoryId",firstCategoryId)
            nObject.put("fourthCategoryId",fourthCategoryId)
            nObject.put("goodUrl",goodUrl)
            nObject.put("good_id",good_id)
            nObject.put("group_num",group_num)
            nObject.put("group_price",group_price)
            nObject.put("is_oversea",is_oversea)
            nObject.put("name",name)
            nObject.put("originProduction",originProduction)
            nObject.put("original_cost",original_cost)
            nObject.put("phone",phone)
            nObject.put("platformId",platformId)
            nObject.put("platformName",platformName)
            nObject.put("priceText",priceText)
            nObject.put("products",products)
            nObject.put("region_id",region_id)
            nObject.put("rootCategoryId",rootCategoryId)
            nObject.put("rootCategoryName",rootCategoryName)
            nObject.put("salesAmount",salesAmount)
            nObject.put("secondCategoryId",secondCategoryId)
            nObject.put("sellCount",sellCount)
            nObject.put("serviceCommitment",serviceCommitment)
            nObject.put("shopId",shopId)
            nObject.put("shopName",shopName)
            nObject.put("shopSales",shopSales)
            nObject.put("shopType",shopType)
            nObject.put("shopUrl",shopUrl)
            nObject.put("subCategoryId",subCategoryId)
            nObject.put("subCategoryName",subCategoryName)
            nObject.put("thirdCategoryId",thirdCategoryId)
            nObject.put("timeStamp",timeStamp)
            nObject.put("title",title)

            nObject.toJSONString
          })

        spark.read.json(value).registerTempTable("frame")

        val data = spark.sql(
            """
              |select
              |
              |allItemCount,
              |categoryId,
              |categoryName,
              |commentCount,
              |crawl_date,
              |emotionalKeywords,
              |evaluates,
              |firstCategoryId,
              |fourthCategoryId,
              |goodUrl,
              |good_id,
              |group_num,
              |group_price,
              |is_oversea,
              |name,
              |originProduction,
              |original_cost,
              |phone,
              |platformId,
              |platformName,
              |priceText,
              |products,
              |region_id,
              |rootCategoryId,
              |rootCategoryName,
              |salesAmount,
              |secondCategoryId,
              |sellCount,
              |serviceCommitment,
              |shopId,
              |shopName,
              |shopSales,
              |shopType,
              |shopUrl,
              |subCategoryId,
              |subCategoryName,
              |thirdCategoryId,
              |timeStamp,
              |title
              |
              |from
              |frame
            """.stripMargin).dropDuplicates("good_id")
        data.repartition(32).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

        val shop = spark.sql(
            """
              |select
              |
              |timeStamp,
              |address,
              |administrative_region,
              |city,
              |city_grade,
              |city_origin,
              |district,
              |district_origin,
              |economic_division,
              |if_city,
              |if_district,
              |if_state_level_new_areas,
              |latitude,
              |longitude,
              |poor_counties,
              |province,
              |regional_ID,
              |rural_demonstration_counties,
              |rural_ecommerce,
              |shopId,
              |the_belt_and_road_city,
              |the_belt_and_road_province,
              |the_yangtze_river_economic_zone_city,
              |the_yangtze_river_economic_zone_province,
              |urban_agglomerations
              |
              |from
              |frame
            """.stripMargin).dropDuplicates("shopId")

        shop.repartition(32).write.orc(s"s3a://dws-data/split/split_addressTable/${year}/${platform}/${month}")

      }
    }
  }
  //美团团购
  def ES_ORC_MTTG(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {
      val frame = spark.read.orc(s"s3a://dws-data/g_data/${year}/${month}/meituan_tg/*")
      if(frame != null){

        val value = frame.toJSON.rdd.map(line => {
          val nObject = JSON.parseObject(line)

          val average_consume = nObject.getOrDefault("average_consume", "-1").toString
          val backCateName = nObject.getOrDefault("backCateName", "-1").toString
          val buildShopId = nObject.getOrDefault("buildShopId", "-1").toString
          val categoryId = nObject.getOrDefault("categoryId", "-1").toString
          val categoryName = nObject.getOrDefault("categoryName", "-1").toString
          val crawl_date = nObject.getOrDefault("crawl_date", "-1").toString
          val dpShopId = nObject.getOrDefault("dpShopId", "-1").toString
          val emotionalKeywords = nObject.getOrDefault("emotionalKeywords", "-1").toString
          val evaluates = nObject.getOrDefault("evaluates", "-1").toString
          val firstCategoryId = nObject.getOrDefault("firstCategoryId", "-1").toString
          val flavors = nObject.getOrDefault("flavors", "-1").toString
          val food_type = nObject.getOrDefault("food_type", "-1").toString
          val fourthCategoryId = nObject.getOrDefault("fourthCategoryId", "-1").toString
          val goodRatePercentage = nObject.getOrDefault("goodRatePercentage", "-1").toString
          val good_id = nObject.getOrDefault("good_id", "-1").toString
          val is_tuan = nObject.getOrDefault("is_tuan", "-1").toString
          val loc_info = nObject.getOrDefault("loc_info", "-1").toString
          val opening_hours = nObject.getOrDefault("opening_hours", "-1").toString
          val original_cost = nObject.getOrDefault("original_cost", "-1").toString
          val phone = nObject.getOrDefault("phone", "-1").toString
          val platformId = nObject.getOrDefault("platformId", "-1").toString
          val platformName = nObject.getOrDefault("platformName", "-1").toString
          val priceText = nObject.getOrDefault("priceText", "-1").toString
          val rootCategoryId = nObject.getOrDefault("rootCategoryId", "-1").toString
          val rootCategoryName = nObject.getOrDefault("rootCategoryName", "-1").toString
          val salesAmount = nObject.getOrDefault("salesAmount", "-1").toString
          val secondCategoryId = nObject.getOrDefault("secondCategoryId", "-1").toString
          val sellCount = nObject.getOrDefault("sellCount", "-1").toString
          val shopAvgscore = nObject.getOrDefault("shopAvgscore", "-1").toString
          val shopCommentCount = nObject.getOrDefault("shopCommentCount", "-1").toString
          val shopImages = nObject.getOrDefault("shopImages", "-1").toString
          val shopName = nObject.getOrDefault("shopName", "-1").toString
          val shopSellCount = nObject.getOrDefault("shopSellCount", "-1").toString
          val shopUrl = nObject.getOrDefault("shopUrl", "-1").toString
          val thirdCategoryId = nObject.getOrDefault("thirdCategoryId", "-1").toString
          val timeStamp = nObject.getOrDefault("timeStamp", "-1").toString
          val title = nObject.getOrDefault("title", "-1").toString
          val shopId = nObject.getOrDefault("shopId", "-1").toString


          nObject.put("average_consume", average_consume)
          nObject.put("shopId", shopId)
          nObject.put("backCateName", backCateName)
          nObject.put("buildShopId", buildShopId)
          nObject.put("categoryId", categoryId)
          nObject.put("categoryName", categoryName)
          nObject.put("crawl_date", crawl_date)
          nObject.put("dpShopId", dpShopId)
          nObject.put("emotionalKeywords", emotionalKeywords)
          nObject.put("evaluates", evaluates)
          nObject.put("firstCategoryId", firstCategoryId)
          nObject.put("flavors", flavors)
          nObject.put("food_type", food_type)
          nObject.put("fourthCategoryId", fourthCategoryId)
          nObject.put("goodRatePercentage", goodRatePercentage)
          nObject.put("good_id", good_id)
          nObject.put("is_tuan", is_tuan)
          nObject.put("loc_info", loc_info)
          nObject.put("opening_hours", opening_hours)
          nObject.put("original_cost", original_cost)
          nObject.put("phone", phone)
          nObject.put("platformId", platformId)
          nObject.put("platformName", platformName)
          nObject.put("priceText", priceText)
          nObject.put("rootCategoryId", rootCategoryId)
          nObject.put("rootCategoryName", rootCategoryName)
          nObject.put("salesAmount", salesAmount)
          nObject.put("secondCategoryId", secondCategoryId)
          nObject.put("sellCount", sellCount)
          nObject.put("shopAvgscore", shopAvgscore)
          nObject.put("shopCommentCount", shopCommentCount)
          nObject.put("shopImages", shopImages)
          nObject.put("shopName", shopName)
          nObject.put("shopSellCount", shopSellCount)
          nObject.put("shopUrl", shopUrl)
          nObject.put("thirdCategoryId", thirdCategoryId)
          nObject.put("timeStamp", timeStamp)
          nObject.put("title", title)


          nObject.toJSONString
        })
        spark.read.json(value).registerTempTable("frame")

        val data = spark.sql(
          """
            |select
            |
            |average_consume,
            |backCateName,
            |buildShopId,
            |categoryId,
            |categoryName,
            |crawl_date,
            |dpShopId,
            |emotionalKeywords,
            |evaluates,
            |firstCategoryId,
            |flavors,
            |food_type,
            |fourthCategoryId,
            |goodRatePercentage,
            |good_id,
            |is_tuan,
            |loc_info,
            |opening_hours,
            |original_cost,
            |phone,
            |platformId,
            |platformName,
            |priceText,
            |rootCategoryId,
            |rootCategoryName,
            |salesAmount,
            |secondCategoryId,
            |sellCount,
            |shopAvgscore,
            |shopCommentCount,
            |shopImages,
            |shopName,
            |shopSellCount,
            |shopUrl,
            |thirdCategoryId,
            |timeStamp,
            |shopId,
            |title
            |
            |from
            |frame
          """.stripMargin).dropDuplicates("good_id")
        data.repartition(32).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

        val shop =  spark.sql(
          """
            |select
            |timeStamp,
            |address,
            |administrative_region,
            |aedzId,
            |city,
            |city_grade,
            |city_origin,
            |economic_division,
            |if_city,
            |if_district,
            |if_state_level_new_areas,
            |latitude,
            |longitude,
            |poor_counties,
            |province,
            |rural_demonstration_counties,
            |rural_ecommerce,
            |shopId,
            |the_belt_and_road_city,
            |the_belt_and_road_province,
            |the_yangtze_river_economic_zone_city,
            |the_yangtze_river_economic_zone_province,
            |town,
            |urban_agglomerations,
            |regional_ID,
            |district,
            |district_origin
            |
            |from
            |frame
          """.stripMargin).dropDuplicates("shopId")
        shop.repartition(32).write.orc(s"s3a://dws-data/split/split_addressTable/${year}/${platform}/${month}")
      }
    }
  }
  //大众点评
  def ES_ORC_DZDP(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {
      val frame = spark.read.orc(s"s3a://dws-data/g_data/${year}/${month}/dazhongdp/*")
      if(frame != null){
        val value = frame.toJSON.rdd.map(line => {
          val nObject = JSON.parseObject(line)

          val average_consume = nObject.getOrDefault("average_consume","-1").toString
          val categoryId = nObject.getOrDefault("categoryId","-1").toString
          val categoryName = nObject.getOrDefault("categoryName","-1").toString
          val crawl_date = nObject.getOrDefault("crawl_date","-1").toString
          val emotionalKeywords = nObject.getOrDefault("emotionalKeywords","-1").toString
          val evaluates = nObject.getOrDefault("evaluates","-1").toString
          val firstCategoryId = nObject.getOrDefault("firstCategoryId","-1").toString
          val flavors = nObject.getOrDefault("flavors","-1").toString
          val food_type = nObject.getOrDefault("food_type","-1").toString
          val fourthCategoryId = nObject.getOrDefault("fourthCategoryId","-1").toString
          val goodRatePercentage = nObject.getOrDefault("goodRatePercentage","-1").toString
          val goodUrl = nObject.getOrDefault("goodUrl","-1").toString
          val good_id = nObject.getOrDefault("good_id","-1").toString
          val is_tuan = nObject.getOrDefault("is_tuan","-1").toString
          val loc_info = nObject.getOrDefault("loc_info","-1").toString
          val opening_hours = nObject.getOrDefault("opening_hours","-1").toString
          val original_cost = nObject.getOrDefault("original_cost","-1").toString
          val phone = nObject.getOrDefault("phone","-1").toString
          val platformId = nObject.getOrDefault("platformId","-1").toString
          val platformName = nObject.getOrDefault("platformName","-1").toString
          val priceText = nObject.getOrDefault("priceText","-1").toString
          val rootCategoryId = nObject.getOrDefault("rootCategoryId","-1").toString
          val rootCategoryName = nObject.getOrDefault("rootCategoryName","-1").toString
          val salesAmount = nObject.getOrDefault("salesAmount","-1").toString
          val secondCategoryId = nObject.getOrDefault("secondCategoryId","-1").toString
          val sellCount = nObject.getOrDefault("sellCount","-1").toString
          val shopCommentCount = nObject.getOrDefault("shopCommentCount","-1").toString
          val shopId = nObject.getOrDefault("shopId","-1").toString
          val shopImages = nObject.getOrDefault("shopImages","-1").toString
          val shopName = nObject.getOrDefault("shopName","-1").toString
          val shopUrl = nObject.getOrDefault("shopUrl","-1").toString
          val thirdCategoryId = nObject.getOrDefault("thirdCategoryId","-1").toString
          val timeStamp = nObject.getOrDefault("timeStamp","-1").toString
          val title = nObject.getOrDefault("title","-1").toString
          val town = nObject.getOrDefault("town", "-1").toString
          val aedzId = nObject.getOrDefault("aedzId", "-1").toString


          nObject.put("town", town)
          nObject.put("aedzId", aedzId)
          nObject.put("average_consume",average_consume)
          nObject.put("categoryId",categoryId)
          nObject.put("categoryName",categoryName)
          nObject.put("crawl_date",crawl_date)
          nObject.put("emotionalKeywords",emotionalKeywords)
          nObject.put("evaluates",evaluates)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("flavors",flavors)
          nObject.put("food_type",food_type)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("goodRatePercentage",goodRatePercentage)
          nObject.put("goodUrl",goodUrl)
          nObject.put("good_id",good_id)
          nObject.put("is_tuan",is_tuan)
          nObject.put("loc_info",loc_info)
          nObject.put("opening_hours",opening_hours)
          nObject.put("original_cost",original_cost)
          nObject.put("phone",phone)
          nObject.put("platformId",platformId)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("salesAmount",salesAmount)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("sellCount",sellCount)
          nObject.put("shopCommentCount",shopCommentCount)
          nObject.put("shopId",shopId)
          nObject.put("shopImages",shopImages)
          nObject.put("shopName",shopName)
          nObject.put("shopUrl",shopUrl)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("title",title)

          nObject.toJSONString
        })

        spark.read.json(value).registerTempTable("frame")

        val data = spark.sql(
          """
            |select
            |
            |average_consume,
            |categoryId,
            |categoryName,
            |crawl_date,
            |emotionalKeywords,
            |evaluates,
            |firstCategoryId,
            |flavors,
            |food_type,
            |fourthCategoryId,
            |goodRatePercentage,
            |goodUrl,
            |good_id,
            |is_tuan,
            |loc_info,
            |opening_hours,
            |original_cost,
            |phone,
            |platformId,
            |platformName,
            |priceText,
            |rootCategoryId,
            |rootCategoryName,
            |salesAmount,
            |secondCategoryId,
            |sellCount,
            |shopCommentCount,
            |shopId,
            |shopImages,
            |shopName,
            |shopUrl,
            |thirdCategoryId,
            |timeStamp,
            |title
            |
            |from
            |frame
          """.stripMargin).dropDuplicates("good_id")

        data.repartition(6).write.orc(s"s3a://dws-data/split/split_data/${year}/dzdp/${month}")

        val shop = spark.sql(
          """
            |select
            |
            |timeStamp,
            |address,
            |administrative_region,
            |aedzId,
            |city,
            |city_grade,
            |city_origin,
            |district,
            |district_origin,
            |economic_division,
            |if_city,
            |if_district,
            |if_state_level_new_areas,
            |latitude,
            |longitude,
            |poor_counties,
            |province,
            |regional_ID,
            |rural_demonstration_counties,
            |rural_ecommerce,
            |shopId,
            |the_belt_and_road_city,
            |the_belt_and_road_province,
            |the_yangtze_river_economic_zone_city,
            |the_yangtze_river_economic_zone_province,
            |town,
            |urban_agglomerations
            |
            |from
            |frame
          """.stripMargin).dropDuplicates("shopId")

        shop.repartition(6).write.orc(s"s3a://dws-data/split/split_addressTable/${year}/dzdp/${month}")
      }
    }
  }

//=============================各平台的提取方法======================================================//
//1、提取京东到家
def ES_ORC_JDDJ(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

  val sc = spark.sparkContext

  for (month <- monthArr) {

    val values = sc.esJsonRDD(s"${year}_jddaojia/jddaojia_${year}_${month}").values

    if(!values.isEmpty()){
      val frame = spark.read.json(values)

      val value = frame.toJSON.rdd.map(line => {
        val nObject = JSON.parseObject(line)
        val obj = new JSONObject()

        val   county  =    nObject.getOrDefault("county","-1").toString
        val   fourthCategoryId  =    nObject.getOrDefault("fourthCategoryId","-1").toString
        val   good_id  =    nObject.getOrDefault("good_id","-1").toString
        val   jd_categoryName  =    nObject.getOrDefault("jd_categoryName","-1").toString
        val   urban_agglomerations  =    nObject.getOrDefault("urban_agglomerations","-1").toString
        val   economic_division  =    nObject.getOrDefault("economic_division","-1").toString
        val   province  =    nObject.getOrDefault("province","-1").toString
        val   jd_categoryId  =    nObject.getOrDefault("jd_categoryId","-1").toString
        val   poor_counties  =    nObject.getOrDefault("poor_counties","-1").toString
        val   brandSourceId  =    nObject.getOrDefault("brandSourceId","-1").toString
        val   rural_demonstration_counties  =    nObject.getOrDefault("rural_demonstration_counties","-1").toString
        val   shopId  =    nObject.getOrDefault("shopId","-1").toString
        val   brand_type  =    nObject.getOrDefault("brand_type","-1").toString
        val   longitude  =    nObject.getOrDefault("longitude","-1").toString
        val   registration_institution  =    nObject.getOrDefault("registration_institution","-1").toString
        val   goodRatePercentage  =    nObject.getOrDefault("goodRatePercentage","-1").toString
        val   brandName  =    nObject.getOrDefault("brandName","-1").toString
        val   images  =    nObject.getOrDefault("images","-1").toString
        val   town  =    nObject.getOrDefault("town","-1").toString
        val   city_origin  =    nObject.getOrDefault("city_origin","-1").toString
        val   if_city  =    nObject.getOrDefault("if_city","-1").toString
        val   crawl_date  =    nObject.getOrDefault("crawl_date","-1").toString
        val   sku_id  =    nObject.getOrDefault("sku_id","-1").toString
        val   shopUrl  =    nObject.getOrDefault("shopUrl","-1").toString
        val   jd_subCategoryName  =    nObject.getOrDefault("jd_subCategoryName","-1").toString
        val   shopSellCount  =    nObject.getOrDefault("shopSellCount","-1").toString
        val   aedzId  =    nObject.getOrDefault("aedzId","-1").toString
        val   brandName_en  =    nObject.getOrDefault("brandName_en","-1").toString
        val   rural_ecommerce  =    nObject.getOrDefault("rural_ecommerce","-1").toString
        val   district  =    nObject.getOrDefault("district","-1").toString
        val   district_origin  =    nObject.getOrDefault("district_origin","-1").toString
        val   name  =    nObject.getOrDefault("name","-1").toString
        val   sellCount  =    nObject.getOrDefault("sellCount","-1").toString
        val   shopType  =    nObject.getOrDefault("shopType","-1").toString
        val   the_yangtze_river_economic_zone_province  =    nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
        val   secondCategoryId  =    nObject.getOrDefault("secondCategoryId","-1").toString
        val   city  =    nObject.getOrDefault("city","-1").toString
        val   latitude  =    nObject.getOrDefault("latitude","-1").toString
        val   serviceTimes  =    nObject.getOrDefault("serviceTimes","-1").toString
        val   venderId  =    nObject.getOrDefault("venderId","-1").toString
        val   brand_isLaoZiHao  =    nObject.getOrDefault("brand_isLaoZiHao","-1").toString
        val   shopName  =    nObject.getOrDefault("shopName","-1").toString
        val   if_state_level_new_areas  =    nObject.getOrDefault("if_state_level_new_areas","-1").toString
        val   city_grade  =    nObject.getOrDefault("city_grade","-1").toString
        val   title  =    nObject.getOrDefault("title","-1").toString
        val   categoryName  =    nObject.getOrDefault("categoryName","-1").toString
        val   firstCategoryId  =    nObject.getOrDefault("firstCategoryId","-1").toString
        val   thirdCategoryId  =    nObject.getOrDefault("thirdCategoryId","-1").toString
        val   jd_subCategoryId  =    nObject.getOrDefault("jd_subCategoryId","-1").toString
        val   salesAmount  =    nObject.getOrDefault("salesAmount","-1").toString
        val   street  =    nObject.getOrDefault("street","-1").toString
        val   administrative_region  =    nObject.getOrDefault("administrative_region","-1").toString
        val   platformName  =    nObject.getOrDefault("platformName","-1").toString
        val   priceText  =    nObject.getOrDefault("priceText","-1").toString
        val   if_district  =    nObject.getOrDefault("if_district","-1").toString
        val   brand_state  =    nObject.getOrDefault("brand_state","-1").toString
        val   address  =    nObject.getOrDefault("address","-1").toString
        val   quantity  =    nObject.getOrDefault("quantity","-1").toString
        val   coupon  =    nObject.getOrDefault("coupon","-1").toString
        val   the_yangtze_river_economic_zone_city  =    nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
        val   shopImages  =    nObject.getOrDefault("shopImages","-1").toString
        val   platformId  =    nObject.getOrDefault("platformId","-1").toString
        val   rootCategoryId  =    nObject.getOrDefault("rootCategoryId","-1").toString
        val   timeStamp  =    nObject.getOrDefault("timeStamp","-1").toString
        val   jd_rootCategoryId  =    nObject.getOrDefault("jd_rootCategoryId","-1").toString
        val   jd_rootCategoryName  =    nObject.getOrDefault("jd_rootCategoryName","-1").toString
        val   the_belt_and_road_city  =    nObject.getOrDefault("the_belt_and_road_city","-1").toString
        val   brandValueId  =    nObject.getOrDefault("brandValueId","-1").toString
        val   the_belt_and_road_province  =    nObject.getOrDefault("the_belt_and_road_province","-1").toString
        val   rootCategoryName  =    nObject.getOrDefault("rootCategoryName","-1").toString
        val   regional_id  =    nObject.getOrDefault("regional_id","-1").toString
        val   brandName_cn  =    nObject.getOrDefault("brandName_cn","-1").toString
        val   categoryId  =    nObject.getOrDefault("categoryId","-1").toString



        obj.put("county",county)
        obj.put("fourthCategoryId",fourthCategoryId)
        obj.put("good_id",good_id)
        obj.put("jd_categoryName",jd_categoryName)
        obj.put("urban_agglomerations",urban_agglomerations)
        obj.put("economic_division",economic_division)
        obj.put("province",province)
        obj.put("jd_categoryId",jd_categoryId)
        obj.put("poor_counties",poor_counties)
        obj.put("brandSourceId",brandSourceId)
        obj.put("rural_demonstration_counties",rural_demonstration_counties)
        obj.put("shopId",shopId)
        obj.put("brand_type",brand_type)
        obj.put("longitude",longitude)
        obj.put("registration_institution",registration_institution)
        obj.put("goodRatePercentage",goodRatePercentage)
        obj.put("brandName",brandName)
        obj.put("images",images)
        obj.put("town",town)
        obj.put("city_origin",city_origin)
        obj.put("if_city",if_city)
        obj.put("crawl_date",crawl_date)
        obj.put("sku_id",sku_id)
        obj.put("shopUrl",shopUrl)
        obj.put("jd_subCategoryName",jd_subCategoryName)
        obj.put("shopSellCount",shopSellCount)
        obj.put("aedzId",aedzId)
        obj.put("brandName_en",brandName_en)
        obj.put("rural_ecommerce",rural_ecommerce)
        obj.put("district",district)
        obj.put("district_origin",district_origin)
        obj.put("name",name)
        obj.put("sellCount",sellCount)
        obj.put("shopType",shopType)
        obj.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
        obj.put("secondCategoryId",secondCategoryId)
        obj.put("city",city)
        obj.put("latitude",latitude)
        obj.put("serviceTimes",serviceTimes)
        obj.put("venderId",venderId)
        obj.put("brand_isLaoZiHao",brand_isLaoZiHao)
        obj.put("shopName",shopName)
        obj.put("if_state_level_new_areas",if_state_level_new_areas)
        obj.put("city_grade",city_grade)
        obj.put("title",title)
        obj.put("categoryName",categoryName)
        obj.put("firstCategoryId",firstCategoryId)
        obj.put("thirdCategoryId",thirdCategoryId)
        obj.put("jd_subCategoryId",jd_subCategoryId)
        obj.put("salesAmount",salesAmount)
        obj.put("street",street)
        obj.put("administrative_region",administrative_region)
        obj.put("platformName",platformName)
        obj.put("priceText",priceText)
        obj.put("if_district",if_district)
        obj.put("brand_state",brand_state)
        obj.put("address",address)
        obj.put("quantity",quantity)
        obj.put("coupon",coupon)
        obj.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
        obj.put("shopImages",shopImages)
        obj.put("platformId",platformId)
        obj.put("rootCategoryId",rootCategoryId)
        obj.put("timeStamp",timeStamp)
        obj.put("jd_rootCategoryId",jd_rootCategoryId)
        obj.put("jd_rootCategoryName",jd_rootCategoryName)
        obj.put("the_belt_and_road_city",the_belt_and_road_city)
        obj.put("brandValueId",brandValueId)
        obj.put("the_belt_and_road_province",the_belt_and_road_province)
        obj.put("rootCategoryName",rootCategoryName)
        obj.put("regional_id",regional_id)
        obj.put("brandName_cn",brandName_cn)
        obj.put("categoryId",categoryId)


        obj.toJSONString
      })

      spark.read.json(value).registerTempTable("frame")

      val df_res = spark.sql(
        """
          |select
          |county,
          |fourthCategoryId,
          |good_id,
          |jd_categoryName,
          |urban_agglomerations,
          |economic_division,
          |province,
          |jd_categoryId,
          |poor_counties,
          |brandSourceId,
          |rural_demonstration_counties,
          |shopId,
          |brand_type,
          |longitude,
          |registration_institution,
          |goodRatePercentage,
          |brandName,
          |images,
          |town,
          |city_origin,
          |if_city,
          |crawl_date,
          |sku_id,
          |shopUrl,
          |jd_subCategoryName,
          |shopSellCount,
          |aedzId,
          |brandName_en,
          |rural_ecommerce,
          |district,
          |district_origin,
          |name,
          |sellCount,
          |shopType,
          |the_yangtze_river_economic_zone_province,
          |secondCategoryId,
          |city,
          |latitude,
          |serviceTimes,
          |venderId,
          |brand_isLaoZiHao,
          |shopName,
          |if_state_level_new_areas,
          |city_grade,
          |title,
          |categoryName,
          |firstCategoryId,
          |thirdCategoryId,
          |jd_subCategoryId,
          |salesAmount,
          |street,
          |administrative_region,
          |platformName,
          |priceText,
          |if_district,
          |brand_state,
          |address,
          |quantity,
          |coupon,
          |the_yangtze_river_economic_zone_city,
          |shopImages,
          |platformId,
          |rootCategoryId,
          |timeStamp,
          |jd_rootCategoryId,
          |jd_rootCategoryName,
          |the_belt_and_road_city,
          |brandValueId,
          |the_belt_and_road_province,
          |rootCategoryName,
          |regional_id,
          |brandName_cn,
          |categoryId
          |
          |
          |from
          |frame
        """.stripMargin)

      df_res.repartition(5).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")
      //      df_res.show()
    }
  }
}

  //1、提取贝店
def ES_ORC_BeiDian(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

  val sc = spark.sparkContext

  for (month <- monthArr) {

    val values = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values

    if(!values.isEmpty()){
      val frame = spark.read.json(values)

      val value = frame.toJSON.rdd.map(line => {
        val nObject = JSON.parseObject(line)

        val   is_oversea  =    nObject.getOrDefault("is_oversea","-1").toString
        val   evaluates  =    nObject.getOrDefault("evaluates","-1").toString
        val   fourthCategoryId  =    nObject.getOrDefault("fourthCategoryId","-1").toString
        val   good_id  =    nObject.getOrDefault("good_id","-1").toString
        val   sellCount_end  =    nObject.getOrDefault("sellCount_end","-1").toString
        val   urban_agglomerations  =    nObject.getOrDefault("urban_agglomerations","-1").toString
        val   economic_division  =    nObject.getOrDefault("economic_division","-1").toString
        val   province  =    nObject.getOrDefault("province","-1").toString
        val   poor_counties  =    nObject.getOrDefault("poor_counties","-1").toString
        val   allItemCount  =    nObject.getOrDefault("allItemCount","-1").toString
        val   emotionalKeywords  =    nObject.getOrDefault("emotionalKeywords","-1").toString
        val   product_id  =    nObject.getOrDefault("product_id","-1").toString
        val   rural_demonstration_counties  =    nObject.getOrDefault("rural_demonstration_counties","-1").toString
        val   brandName  =    nObject.getOrDefault("brandName","-1").toString
        val   images  =    nObject.getOrDefault("images","-1").toString
        val   goodUrl  =    nObject.getOrDefault("goodUrl","-1").toString
        val   city_origin  =    nObject.getOrDefault("city_origin","-1").toString
        val   if_city  =    nObject.getOrDefault("if_city","-1").toString
        val   subCategoryId  =    nObject.getOrDefault("subCategoryId","-1").toString
        val   serviceCommitment  =    nObject.getOrDefault("serviceCommitment","-1").toString
        val   is_onsell  =    nObject.getOrDefault("is_onsell","-1").toString
        val   rural_ecommerce  =    nObject.getOrDefault("rural_ecommerce","-1").toString
        val   district  =    nObject.getOrDefault("district","-1").toString
        val   delivery_from  =    nObject.getOrDefault("delivery_from","-1").toString
        val   sellCount  =    nObject.getOrDefault("sellCount","-1").toString
        val   name  =    nObject.getOrDefault("name","-1").toString
        val   district_origin  =    nObject.getOrDefault("district_origin","-1").toString
        val   shopType  =    nObject.getOrDefault("shopType","-1").toString
        val   the_yangtze_river_economic_zone_province  =    nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
        val   secondCategoryId  =    nObject.getOrDefault("secondCategoryId","-1").toString
        val   city  =    nObject.getOrDefault("city","-1").toString
        val   Base_Info  =    nObject.getOrDefault("Base_Info","-1").toString
        val   if_state_level_new_areas  =    nObject.getOrDefault("if_state_level_new_areas","-1").toString
        val   title  =    nObject.getOrDefault("title","-1").toString
        val   city_grade  =    nObject.getOrDefault("city_grade","-1").toString
        val   categoryName  =    nObject.getOrDefault("categoryName","-1").toString
        val   firstCategoryId  =    nObject.getOrDefault("firstCategoryId","-1").toString
        val   thirdCategoryId  =    nObject.getOrDefault("thirdCategoryId","-1").toString
        val   subCategoryName  =    nObject.getOrDefault("subCategoryName","-1").toString
        val   salesAmount  =    nObject.getOrDefault("salesAmount","-1").toString
        val   administrative_region  =    nObject.getOrDefault("administrative_region","-1").toString
        val   platformName  =    nObject.getOrDefault("platformName","-1").toString
        val   priceText  =    nObject.getOrDefault("priceText","-1").toString
        val   if_district  =    nObject.getOrDefault("if_district","-1").toString
        val   address  =    nObject.getOrDefault("address","-1").toString
        val   the_yangtze_river_economic_zone_city  =    nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
        val   originProduction  =    nObject.getOrDefault("originProduction","-1").toString
        val   platformId  =    nObject.getOrDefault("platformId","-1").toString
        val   rootCategoryId  =    nObject.getOrDefault("rootCategoryId","-1").toString
        val   timeStamp  =    nObject.getOrDefault("timeStamp","-1").toString
        val   original_cost  =    nObject.getOrDefault("original_cost","-1").toString
        val   the_belt_and_road_city  =    nObject.getOrDefault("the_belt_and_road_city","-1").toString
        val   brandValueId  =    nObject.getOrDefault("brandValueId","-1").toString
        val   originItem  =    nObject.getOrDefault("originItem","-1").toString
        val   the_belt_and_road_province  =    nObject.getOrDefault("the_belt_and_road_province","-1").toString
        val   rootCategoryName  =    nObject.getOrDefault("rootCategoryName","-1").toString
        val   regional_ID  =    nObject.getOrDefault("regional_ID","-1").toString
        val   categoryId  =    nObject.getOrDefault("categoryId","-1").toString
        val   sellCount_start  =    nObject.getOrDefault("sellCount_start","-1").toString


        nObject.put("is_oversea",is_oversea)
        nObject.put("evaluates",evaluates)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("good_id",good_id)
        nObject.put("sellCount_end",sellCount_end)
        nObject.put("urban_agglomerations",urban_agglomerations)
        nObject.put("economic_division",economic_division)
        nObject.put("province",province)
        nObject.put("poor_counties",poor_counties)
        nObject.put("allItemCount",allItemCount)
        nObject.put("emotionalKeywords",emotionalKeywords)
        nObject.put("product_id",product_id)
        nObject.put("rural_demonstration_counties",rural_demonstration_counties)
        nObject.put("brandName",brandName)
        nObject.put("images",images)
        nObject.put("goodUrl",goodUrl)
        nObject.put("city_origin",city_origin)
        nObject.put("if_city",if_city)
        nObject.put("subCategoryId",subCategoryId)
        nObject.put("serviceCommitment",serviceCommitment)
        nObject.put("is_onsell",is_onsell)
        nObject.put("rural_ecommerce",rural_ecommerce)
        nObject.put("district",district)
        nObject.put("delivery_from",delivery_from)
        nObject.put("sellCount",sellCount)
        nObject.put("name",name)
        nObject.put("district_origin",district_origin)
        nObject.put("shopType",shopType)
        nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("city",city)
        nObject.put("Base_Info",Base_Info)
        nObject.put("if_state_level_new_areas",if_state_level_new_areas)
        nObject.put("title",title)
        nObject.put("city_grade",city_grade)
        nObject.put("categoryName",categoryName)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("subCategoryName",subCategoryName)
        nObject.put("salesAmount",salesAmount)
        nObject.put("administrative_region",administrative_region)
        nObject.put("platformName",platformName)
        nObject.put("priceText",priceText)
        nObject.put("if_district",if_district)
        nObject.put("address",address)
        nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
        nObject.put("originProduction",originProduction)
        nObject.put("platformId",platformId)
        nObject.put("rootCategoryId",rootCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("original_cost",original_cost)
        nObject.put("the_belt_and_road_city",the_belt_and_road_city)
        nObject.put("brandValueId",brandValueId)
        nObject.put("originItem",originItem)
        nObject.put("the_belt_and_road_province",the_belt_and_road_province)
        nObject.put("rootCategoryName",rootCategoryName)
        nObject.put("regional_ID",regional_ID)
        nObject.put("categoryId",categoryId)
        nObject.put("sellCount_start",sellCount_start)

        nObject.toJSONString
      })

      spark.read.json(value).registerTempTable("frame")

      val df_res = spark.sql(
        """
          |select
          |is_oversea,
          |evaluates,
          |fourthCategoryId,
          |good_id,
          |sellCount_end,
          |urban_agglomerations,
          |economic_division,
          |province,
          |poor_counties,
          |allItemCount,
          |emotionalKeywords,
          |product_id,
          |rural_demonstration_counties,
          |brandName,
          |images,
          |goodUrl,
          |city_origin,
          |if_city,
          |subCategoryId,
          |serviceCommitment,
          |is_onsell,
          |rural_ecommerce,
          |district,
          |delivery_from,
          |sellCount,
          |name,
          |district_origin,
          |shopType,
          |the_yangtze_river_economic_zone_province,
          |secondCategoryId,
          |city,
          |Base_Info,
          |if_state_level_new_areas,
          |title,
          |city_grade,
          |categoryName,
          |firstCategoryId,
          |thirdCategoryId,
          |subCategoryName,
          |salesAmount,
          |administrative_region,
          |platformName,
          |priceText,
          |if_district,
          |address,
          |the_yangtze_river_economic_zone_city,
          |originProduction,
          |platformId,
          |rootCategoryId,
          |timeStamp,
          |original_cost,
          |the_belt_and_road_city,
          |brandValueId,
          |originItem,
          |the_belt_and_road_province,
          |rootCategoryName,
          |regional_ID,
          |categoryId,
          |sellCount_start
          |
          |from
          |frame
        """.stripMargin)

      df_res.repartition(5).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")
      //      df_res.show()
    }
  }
}


  //1、提取酒仙网
def ES_ORC_JX(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

  val sc = spark.sparkContext

  for (month <- monthArr) {

    val values = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values

    if(!values.isEmpty()){
      val frame = spark.read.json(values)

      val value = frame.toJSON.rdd.map(line => {
        val nObject = JSON.parseObject(line)

        val   businessUrl  =    nObject.getOrDefault("businessUrl","-1").toString
        val   good_id  =    nObject.getOrDefault("good_id","-1").toString
        val   fourthCategoryId  =    nObject.getOrDefault("fourthCategoryId","-1").toString
        val   urban_agglomerations  =    nObject.getOrDefault("urban_agglomerations","-1").toString
        val   economic_division  =    nObject.getOrDefault("economic_division","-1").toString
        val   province  =    nObject.getOrDefault("province","-1").toString
        val   poor_counties  =    nObject.getOrDefault("poor_counties","-1").toString
        val   emotionalKeywords  =    nObject.getOrDefault("emotionalKeywords","-1").toString
        val   brandSourceId  =    nObject.getOrDefault("brandSourceId","-1").toString
        val   shopId  =    nObject.getOrDefault("shopId","-1").toString
        val   rural_demonstration_counties  =    nObject.getOrDefault("rural_demonstration_counties","-1").toString
        val   brand_type  =    nObject.getOrDefault("brand_type","-1").toString
        val   longitude  =    nObject.getOrDefault("longitude","-1").toString
        val   registration_institution  =    nObject.getOrDefault("registration_institution","-1").toString
        val   brandName  =    nObject.getOrDefault("brandName","-1").toString
        val   images  =    nObject.getOrDefault("images","-1").toString
        val   goodUrl  =    nObject.getOrDefault("goodUrl","-1").toString
        val   town  =    nObject.getOrDefault("town","-1").toString
        val   discount_re4  =    nObject.getOrDefault("discount_re4","-1").toString
        val   city_origin  =    nObject.getOrDefault("city_origin","-1").toString
        val   if_city  =    nObject.getOrDefault("if_city","-1").toString
        val   shopUrl  =    nObject.getOrDefault("shopUrl","-1").toString
        val   serviceCommitment  =    nObject.getOrDefault("serviceCommitment","-1").toString
        val   is_onsell  =    nObject.getOrDefault("is_onsell","-1").toString
        val   aedzId  =    nObject.getOrDefault("aedzId","-1").toString
        val   brandName_en  =    nObject.getOrDefault("brandName_en","-1").toString
        val   rural_ecommerce  =    nObject.getOrDefault("rural_ecommerce","-1").toString
        val   district  =    nObject.getOrDefault("district","-1").toString
        val   promotion_info  =    nObject.getOrDefault("promotion_info","-1").toString
        val   sellCount  =    nObject.getOrDefault("sellCount","-1").toString
        val   district_origin  =    nObject.getOrDefault("district_origin","-1").toString
        val   name  =    nObject.getOrDefault("name","-1").toString
        val   shopType  =    nObject.getOrDefault("shopType","-1").toString
        val   the_yangtze_river_economic_zone_province  =    nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
        val   secondCategoryId  =    nObject.getOrDefault("secondCategoryId","-1").toString
        val   business_info  =    nObject.getOrDefault("business_info","-1").toString
        val   city  =    nObject.getOrDefault("city","-1").toString
        val   Base_Info  =    nObject.getOrDefault("Base_Info","-1").toString
        val   freight  =    nObject.getOrDefault("freight","-1").toString
        val   latitude  =    nObject.getOrDefault("latitude","-1").toString
        val   brand_isLaoZiHao  =    nObject.getOrDefault("brand_isLaoZiHao","-1").toString
        val   shopName  =    nObject.getOrDefault("shopName","-1").toString
        val   if_state_level_new_areas  =    nObject.getOrDefault("if_state_level_new_areas","-1").toString
        val   title  =    nObject.getOrDefault("title","-1").toString
        val   city_grade  =    nObject.getOrDefault("city_grade","-1").toString
        val   firstCategoryId  =    nObject.getOrDefault("firstCategoryId","-1").toString
        val   thirdCategoryId  =    nObject.getOrDefault("thirdCategoryId","-1").toString
        val   salesAmount  =    nObject.getOrDefault("salesAmount","-1").toString
        val   administrative_region  =    nObject.getOrDefault("administrative_region","-1").toString
        val   platformName  =    nObject.getOrDefault("platformName","-1").toString
        val   if_district  =    nObject.getOrDefault("if_district","-1").toString
        val   brand_state  =    nObject.getOrDefault("brand_state","-1").toString
        val   address  =    nObject.getOrDefault("address","-1").toString
        val   coupon  =    nObject.getOrDefault("coupon","-1").toString
        val   the_yangtze_river_economic_zone_city  =    nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
        val   platformId  =    nObject.getOrDefault("platformId","-1").toString
        val   rootCategoryId  =    nObject.getOrDefault("rootCategoryId","-1").toString
        val   timeStamp  =    nObject.getOrDefault("timeStamp","-1").toString
        val   the_belt_and_road_city  =    nObject.getOrDefault("the_belt_and_road_city","-1").toString
        val   brandValueId  =    nObject.getOrDefault("brandValueId","-1").toString
        val   the_belt_and_road_province  =    nObject.getOrDefault("the_belt_and_road_province","-1").toString
        val   rootCategoryName  =    nObject.getOrDefault("rootCategoryName","-1").toString
        val   regional_ID  =    nObject.getOrDefault("regional_ID","-1").toString
        val   brandName_cn  =    nObject.getOrDefault("brandName_cn","-1").toString


        nObject.put("businessUrl",businessUrl)
        nObject.put("good_id",good_id)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("urban_agglomerations",urban_agglomerations)
        nObject.put("economic_division",economic_division)
        nObject.put("province",province)
        nObject.put("poor_counties",poor_counties)
        nObject.put("emotionalKeywords",emotionalKeywords)
        nObject.put("brandSourceId",brandSourceId)
        nObject.put("shopId",shopId)
        nObject.put("rural_demonstration_counties",rural_demonstration_counties)
        nObject.put("brand_type",brand_type)
        nObject.put("longitude",longitude)
        nObject.put("registration_institution",registration_institution)
        nObject.put("brandName",brandName)
        nObject.put("images",images)
        nObject.put("goodUrl",goodUrl)
        nObject.put("town",town)
        nObject.put("discount_re4",discount_re4)
        nObject.put("city_origin",city_origin)
        nObject.put("if_city",if_city)
        nObject.put("shopUrl",shopUrl)
        nObject.put("serviceCommitment",serviceCommitment)
        nObject.put("is_onsell",is_onsell)
        nObject.put("aedzId",aedzId)
        nObject.put("brandName_en",brandName_en)
        nObject.put("rural_ecommerce",rural_ecommerce)
        nObject.put("district",district)
        nObject.put("promotion_info",promotion_info)
        nObject.put("sellCount",sellCount)
        nObject.put("district_origin",district_origin)
        nObject.put("name",name)
        nObject.put("shopType",shopType)
        nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("business_info",business_info)
        nObject.put("city",city)
        nObject.put("Base_Info",Base_Info)
        nObject.put("freight",freight)
        nObject.put("latitude",latitude)
        nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
        nObject.put("shopName",shopName)
        nObject.put("if_state_level_new_areas",if_state_level_new_areas)
        nObject.put("title",title)
        nObject.put("city_grade",city_grade)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("salesAmount",salesAmount)
        nObject.put("administrative_region",administrative_region)
        nObject.put("platformName",platformName)
        nObject.put("if_district",if_district)
        nObject.put("brand_state",brand_state)
        nObject.put("address",address)
        nObject.put("coupon",coupon)
        nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
        nObject.put("platformId",platformId)
        nObject.put("rootCategoryId",rootCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("the_belt_and_road_city",the_belt_and_road_city)
        nObject.put("brandValueId",brandValueId)
        nObject.put("the_belt_and_road_province",the_belt_and_road_province)
        nObject.put("rootCategoryName",rootCategoryName)
        nObject.put("regional_ID",regional_ID)
        nObject.put("brandName_cn",brandName_cn)

        nObject.toJSONString
      })

      spark.read.json(value).registerTempTable("frame")

      val df_res = spark.sql(
        """
          |select
          |businessUrl,
          |good_id,
          |fourthCategoryId,
          |urban_agglomerations,
          |economic_division,
          |province,
          |poor_counties,
          |emotionalKeywords,
          |brandSourceId,
          |shopId,
          |rural_demonstration_counties,
          |brand_type,
          |longitude,
          |registration_institution,
          |brandName,
          |images,
          |goodUrl,
          |town,
          |discount_re4,
          |city_origin,
          |if_city,
          |shopUrl,
          |serviceCommitment,
          |is_onsell,
          |aedzId,
          |brandName_en,
          |rural_ecommerce,
          |district,
          |promotion_info,
          |sellCount,
          |district_origin,
          |name,
          |shopType,
          |the_yangtze_river_economic_zone_province,
          |secondCategoryId,
          |business_info,
          |city,
          |Base_Info,
          |freight,
          |latitude,
          |brand_isLaoZiHao,
          |shopName,
          |if_state_level_new_areas,
          |title,
          |city_grade,
          |firstCategoryId,
          |thirdCategoryId,
          |salesAmount,
          |administrative_region,
          |platformName,
          |if_district,
          |brand_state,
          |address,
          |coupon,
          |the_yangtze_river_economic_zone_city,
          |platformId,
          |rootCategoryId,
          |timeStamp,
          |the_belt_and_road_city,
          |brandValueId,
          |the_belt_and_road_province,
          |rootCategoryName,
          |regional_ID,
          |brandName_cn
          |
          |from
          |frame
        """.stripMargin)

      df_res.repartition(5).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")
      //      df_res.show()
    }
  }
}
  //1、提取融易购
  def ES_ORC_RYG(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext

    for (month <- monthArr) {

      val values = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values

      if(!values.isEmpty()){
        val frame = spark.read.json(values)

        val value = frame.toJSON.rdd.map(line => {
          val nObject = JSON.parseObject(line)

          val   is_oversea  =    nObject.getOrDefault("is_oversea","-1").toString
          val   evaluates  =    nObject.getOrDefault("evaluates","-1").toString
          val   fourthCategoryId  =    nObject.getOrDefault("fourthCategoryId","-1").toString
          val   good_id  =    nObject.getOrDefault("good_id","-1").toString
          val   urban_agglomerations  =    nObject.getOrDefault("urban_agglomerations","-1").toString
          val   economic_division  =    nObject.getOrDefault("economic_division","-1").toString
          val   province  =    nObject.getOrDefault("province","-1").toString
          val   poor_counties  =    nObject.getOrDefault("poor_counties","-1").toString
          val   brandSourceId  =    nObject.getOrDefault("brandSourceId","-1").toString
          val   shopId  =    nObject.getOrDefault("shopId","-1").toString
          val   rural_demonstration_counties  =    nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val   brand_type  =    nObject.getOrDefault("brand_type","-1").toString
          val   longitude  =    nObject.getOrDefault("longitude","-1").toString
          val   registration_institution  =    nObject.getOrDefault("registration_institution","-1").toString
          val   goodRatePercentage  =    nObject.getOrDefault("goodRatePercentage","-1").toString
          val   brandName  =    nObject.getOrDefault("brandName","-1").toString
          val   images  =    nObject.getOrDefault("images","-1").toString
          val   goodUrl  =    nObject.getOrDefault("goodUrl","-1").toString
          val   town  =    nObject.getOrDefault("town","-1").toString
          val   city_origin  =    nObject.getOrDefault("city_origin","-1").toString
          val   if_city  =    nObject.getOrDefault("if_city","-1").toString
          val   subCategoryId  =    nObject.getOrDefault("subCategoryId","-1").toString
          val   shopUrl  =    nObject.getOrDefault("shopUrl","-1").toString
          val   aedzId  =    nObject.getOrDefault("aedzId","-1").toString
          val   brandName_en  =    nObject.getOrDefault("brandName_en","-1").toString
          val   rural_ecommerce  =    nObject.getOrDefault("rural_ecommerce","-1").toString
          val   company_name  =    nObject.getOrDefault("company_name","-1").toString
          val   district  =    nObject.getOrDefault("district","-1").toString
          val   sellCount  =    nObject.getOrDefault("sellCount","-1").toString
          val   district_origin  =    nObject.getOrDefault("district_origin","-1").toString
          val   name  =    nObject.getOrDefault("name","-1").toString
          val   shopType  =    nObject.getOrDefault("shopType","-1").toString
          val   the_yangtze_river_economic_zone_province  =    nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val   secondCategoryId  =    nObject.getOrDefault("secondCategoryId","-1").toString
          val   city  =    nObject.getOrDefault("city","-1").toString
          val   Base_Info  =    nObject.getOrDefault("Base_Info","-1").toString
          val   latitude  =    nObject.getOrDefault("latitude","-1").toString
          val   brand_isLaoZiHao  =    nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val   shopName  =    nObject.getOrDefault("shopName","-1").toString
          val   if_state_level_new_areas  =    nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val   title  =    nObject.getOrDefault("title","-1").toString
          val   city_grade  =    nObject.getOrDefault("city_grade","-1").toString
          val   categoryName  =    nObject.getOrDefault("categoryName","-1").toString
          val   firstCategoryId  =    nObject.getOrDefault("firstCategoryId","-1").toString
          val   thirdCategoryId  =    nObject.getOrDefault("thirdCategoryId","-1").toString
          val   subCategoryName  =    nObject.getOrDefault("subCategoryName","-1").toString
          val   goodCount  =    nObject.getOrDefault("goodCount","-1").toString
          val   salesAmount  =    nObject.getOrDefault("salesAmount","-1").toString
          val   administrative_region  =    nObject.getOrDefault("administrative_region","-1").toString
          val   platformName  =    nObject.getOrDefault("platformName","-1").toString
          val   priceText  =    nObject.getOrDefault("priceText","-1").toString
          val   poorCount  =    nObject.getOrDefault("poorCount","-1").toString
          val   if_district  =    nObject.getOrDefault("if_district","-1").toString
          val   brand_state  =    nObject.getOrDefault("brand_state","-1").toString
          val   quantity  =    nObject.getOrDefault("quantity","-1").toString
          val   address  =    nObject.getOrDefault("address","-1").toString
          val   the_yangtze_river_economic_zone_city  =    nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val   originProduction  =    nObject.getOrDefault("originProduction","-1").toString
          val   platformId  =    nObject.getOrDefault("platformId","-1").toString
          val   rootCategoryId  =    nObject.getOrDefault("rootCategoryId","-1").toString
          val   timeStamp  =    nObject.getOrDefault("timeStamp","-1").toString
          val   original_cost  =    nObject.getOrDefault("original_cost","-1").toString
          val   generalCount  =    nObject.getOrDefault("generalCount","-1").toString
          val   the_belt_and_road_city  =    nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val   brandValueId  =    nObject.getOrDefault("brandValueId","-1").toString
          val   originItem  =    nObject.getOrDefault("originItem","-1").toString
          val   the_belt_and_road_province  =    nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val   rootCategoryName  =    nObject.getOrDefault("rootCategoryName","-1").toString
          val   regional_ID  =    nObject.getOrDefault("regional_ID","-1").toString
          val   brandName_cn  =    nObject.getOrDefault("brandName_cn","-1").toString
          val   categoryId  =    nObject.getOrDefault("categoryId","-1").toString


          nObject.put("is_oversea",is_oversea)
          nObject.put("evaluates",evaluates)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("good_id",good_id)
          nObject.put("urban_agglomerations",urban_agglomerations)
          nObject.put("economic_division",economic_division)
          nObject.put("province",province)
          nObject.put("poor_counties",poor_counties)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("shopId",shopId)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("brand_type",brand_type)
          nObject.put("longitude",longitude)
          nObject.put("registration_institution",registration_institution)
          nObject.put("goodRatePercentage",goodRatePercentage)
          nObject.put("brandName",brandName)
          nObject.put("images",images)
          nObject.put("goodUrl",goodUrl)
          nObject.put("town",town)
          nObject.put("city_origin",city_origin)
          nObject.put("if_city",if_city)
          nObject.put("subCategoryId",subCategoryId)
          nObject.put("shopUrl",shopUrl)
          nObject.put("aedzId",aedzId)
          nObject.put("brandName_en",brandName_en)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("company_name",company_name)
          nObject.put("district",district)
          nObject.put("sellCount",sellCount)
          nObject.put("district_origin",district_origin)
          nObject.put("name",name)
          nObject.put("shopType",shopType)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("city",city)
          nObject.put("Base_Info",Base_Info)
          nObject.put("latitude",latitude)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("shopName",shopName)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("title",title)
          nObject.put("city_grade",city_grade)
          nObject.put("categoryName",categoryName)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("subCategoryName",subCategoryName)
          nObject.put("goodCount",goodCount)
          nObject.put("salesAmount",salesAmount)
          nObject.put("administrative_region",administrative_region)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("poorCount",poorCount)
          nObject.put("if_district",if_district)
          nObject.put("brand_state",brand_state)
          nObject.put("quantity",quantity)
          nObject.put("address",address)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("originProduction",originProduction)
          nObject.put("platformId",platformId)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("original_cost",original_cost)
          nObject.put("generalCount",generalCount)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("brandValueId",brandValueId)
          nObject.put("originItem",originItem)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("regional_ID",regional_ID)
          nObject.put("brandName_cn",brandName_cn)
          nObject.put("categoryId",categoryId)

          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          nObject.toJSONString
        })

        spark.read.json(value).registerTempTable("frame")

        val df_res = spark.sql(
          """
            |select
            |is_oversea,
            |evaluates,
            |fourthCategoryId,
            |good_id,
            |urban_agglomerations,
            |economic_division,
            |province,
            |poor_counties,
            |brandSourceId,
            |shopId,
            |rural_demonstration_counties,
            |brand_type,
            |longitude,
            |registration_institution,
            |goodRatePercentage,
            |brandName,
            |images,
            |goodUrl,
            |town,
            |city_origin,
            |if_city,
            |subCategoryId,
            |shopUrl,
            |aedzId,
            |brandName_en,
            |rural_ecommerce,
            |company_name,
            |district,
            |sellCount,
            |district_origin,
            |name,
            |shopType,
            |the_yangtze_river_economic_zone_province,
            |secondCategoryId,
            |city,
            |Base_Info,
            |latitude,
            |brand_isLaoZiHao,
            |shopName,
            |if_state_level_new_areas,
            |title,
            |city_grade,
            |categoryName,
            |firstCategoryId,
            |thirdCategoryId,
            |subCategoryName,
            |goodCount,
            |salesAmount,
            |administrative_region,
            |platformName,
            |priceText,
            |poorCount,
            |if_district,
            |brand_state,
            |quantity,
            |address,
            |the_yangtze_river_economic_zone_city,
            |originProduction,
            |platformId,
            |rootCategoryId,
            |timeStamp,
            |original_cost,
            |generalCount,
            |the_belt_and_road_city,
            |brandValueId,
            |originItem,
            |the_belt_and_road_province,
            |rootCategoryName,
            |regional_ID,
            |brandName_cn,
            |categoryId
            |
            |from
            |frame
          """.stripMargin)

        df_res.repartition(5).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")
        //      df_res.show()
      }
    }
  }
  //2、提取beibei
  def ES_OBS_2_beibei(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {
        val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
        var t1 = value.map(x => {
          val nObject = JSON.parseObject(x)

          //1、将所有的字段转为string类型
          val  brandallItemCount =  nObject.getOrDefault("brandallItemCount","-1").toString
          val  is_oversea =  nObject.getOrDefault("is_oversea","-1").toString
          val  evaluates =  nObject.getOrDefault("evaluates","-1").toString
          val  fourthCategoryId =  nObject.getOrDefault("fourthCategoryId","-1").toString
          val  good_id =  nObject.getOrDefault("good_id","-1").toString
          val  urban_agglomerations =  nObject.getOrDefault("urban_agglomerations","-1").toString
          val  economic_division =  nObject.getOrDefault("economic_division","-1").toString
          val  score =  nObject.getOrDefault("score","-1").toString
          val  province =  nObject.getOrDefault("province","-1").toString
          val  poor_counties =  nObject.getOrDefault("poor_counties","-1").toString
          val  sellCount_0601 =  nObject.getOrDefault("sellCount_0601","-1").toString
          val  allItemCount =  nObject.getOrDefault("allItemCount","-1").toString
          val  brandSourceId =  nObject.getOrDefault("brandSourceId","-1").toString
          val  rural_demonstration_counties =  nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val  shopId =  nObject.getOrDefault("shopId","-1").toString
          val  brand_type =  nObject.getOrDefault("brand_type","-1").toString
          val  brandFavCount =  nObject.getOrDefault("brandFavCount","-1").toString
          val  goodRatePercentage =  nObject.getOrDefault("goodRatePercentage","-1").toString
          val  brandName =  nObject.getOrDefault("brandName","-1").toString
          val  images =  nObject.getOrDefault("images","-1").toString
          val  goodUrl =  nObject.getOrDefault("goodUrl","-1").toString
          val  brandevaluates =  nObject.getOrDefault("brandevaluates","-1").toString
          val  city_origin =  nObject.getOrDefault("city_origin","-1").toString
          val  if_city =  nObject.getOrDefault("if_city","-1").toString
          val  crawl_date =  nObject.getOrDefault("crawl_date","-1").toString
          val  subCategoryId =  nObject.getOrDefault("subCategoryId","-1").toString
          val  shopUrl =  nObject.getOrDefault("shopUrl","-1").toString
          val  is_onsell =  nObject.getOrDefault("is_onsell","-1").toString
          val  brandName_en =  nObject.getOrDefault("brandName_en","-1").toString
          val  rural_ecommerce =  nObject.getOrDefault("rural_ecommerce","-1").toString
          val  district =  nObject.getOrDefault("district","-1").toString
          val  taxFreeZone =  nObject.getOrDefault("taxFreeZone","-1").toString
          val  delivery_from =  nObject.getOrDefault("delivery_from","-1").toString
          val  district_origin =  nObject.getOrDefault("district_origin","-1").toString
          val  name =  nObject.getOrDefault("name","-1").toString
          val  sellCount =  nObject.getOrDefault("sellCount","-1").toString
          val  shopType =  nObject.getOrDefault("shopType","-1").toString
          val  the_yangtze_river_economic_zone_province =  nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val  secondCategoryId =  nObject.getOrDefault("secondCategoryId","-1").toString
          val  city =  nObject.getOrDefault("city","-1").toString
          val  Base_Info =  nObject.getOrDefault("Base_Info","-1").toString
          val  brand_isLaoZiHao =  nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val  shopName =  nObject.getOrDefault("shopName","-1").toString
          val  if_state_level_new_areas =  nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val  city_grade =  nObject.getOrDefault("city_grade","-1").toString
          val  title =  nObject.getOrDefault("title","-1").toString
          val  categoryName =  nObject.getOrDefault("categoryName","-1").toString
          val  firstCategoryId =  nObject.getOrDefault("firstCategoryId","-1").toString
          val  thirdCategoryId =  nObject.getOrDefault("thirdCategoryId","-1").toString
          val  subCategoryName =  nObject.getOrDefault("subCategoryName","-1").toString
          val  salesAmount =  nObject.getOrDefault("salesAmount","-1").toString
          val  administrative_region =  nObject.getOrDefault("administrative_region","-1").toString
          val  platformName =  nObject.getOrDefault("platformName","-1").toString
          val  priceText =  nObject.getOrDefault("priceText","-1").toString
          val  if_district =  nObject.getOrDefault("if_district","-1").toString
          val  brand_state =  nObject.getOrDefault("brand_state","-1").toString
          val  address =  nObject.getOrDefault("address","-1").toString
          val  the_yangtze_river_economic_zone_city =  nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val  platformId =  nObject.getOrDefault("platformId","-1").toString
          val  fans =  nObject.getOrDefault("fans","-1").toString
          val  sellCount_0618 =  nObject.getOrDefault("sellCount_0618","-1").toString
          val  timeStamp =  nObject.getOrDefault("timeStamp","-1").toString
          val  original_cost =  nObject.getOrDefault("original_cost","-1").toString
          val  the_belt_and_road_city =  nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val  brandValueId =  nObject.getOrDefault("brandValueId","-1").toString
          val  originItem =  nObject.getOrDefault("originItem","-1").toString
          val  the_belt_and_road_province =  nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val  originCountryName =  nObject.getOrDefault("originCountryName","-1").toString
          val  is_selfSupport =  nObject.getOrDefault("is_selfSupport","-1").toString
          val  rootCategoryName =  nObject.getOrDefault("rootCategoryName","-1").toString
          val  regional_ID =  nObject.getOrDefault("regional_ID","-1").toString
          val  brandName_cn =  nObject.getOrDefault("brandName_cn","-1").toString
          val  customCategoryId =  nObject.getOrDefault("customCategoryId","-1").toString


          nObject.put("brandallItemCount",brandallItemCount)
          nObject.put("is_oversea",is_oversea)
          nObject.put("evaluates",evaluates)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("good_id",good_id)
          nObject.put("urban_agglomerations",urban_agglomerations)
          nObject.put("economic_division",economic_division)
          nObject.put("score",score)
          nObject.put("province",province)
          nObject.put("poor_counties",poor_counties)
          nObject.put("sellCount_0601",sellCount_0601)
          nObject.put("allItemCount",allItemCount)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("shopId",shopId)
          nObject.put("brand_type",brand_type)
          nObject.put("brandFavCount",brandFavCount)
          nObject.put("goodRatePercentage",goodRatePercentage)
          nObject.put("brandName",brandName)
          nObject.put("images",images)
          nObject.put("goodUrl",goodUrl)
          nObject.put("brandevaluates",brandevaluates)
          nObject.put("city_origin",city_origin)
          nObject.put("if_city",if_city)
          nObject.put("crawl_date",crawl_date)
          nObject.put("subCategoryId",subCategoryId)
          nObject.put("shopUrl",shopUrl)
          nObject.put("is_onsell",is_onsell)
          nObject.put("brandName_en",brandName_en)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("district",district)
          nObject.put("taxFreeZone",taxFreeZone)
          nObject.put("delivery_from",delivery_from)
          nObject.put("district_origin",district_origin)
          nObject.put("name",name)
          nObject.put("sellCount",sellCount)
          nObject.put("shopType",shopType)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("city",city)
          nObject.put("Base_Info",Base_Info)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("shopName",shopName)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("city_grade",city_grade)
          nObject.put("title",title)
          nObject.put("categoryName",categoryName)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("subCategoryName",subCategoryName)
          nObject.put("salesAmount",salesAmount)
          nObject.put("administrative_region",administrative_region)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("if_district",if_district)
          nObject.put("brand_state",brand_state)
          nObject.put("address",address)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("platformId","1")
          nObject.put("fans",fans)
          nObject.put("sellCount_0618",sellCount_0618)
          nObject.put("timeStamp",timeStamp)
          nObject.put("original_cost",original_cost)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("brandValueId",brandValueId)
          nObject.put("originItem",originItem)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("originCountryName",originCountryName)
          nObject.put("is_selfSupport",is_selfSupport)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("regional_ID",regional_ID)
          nObject.put("brandName_cn",brandName_cn)
          nObject.put("customCategoryId",customCategoryId)


          //2、去掉某些字段
          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          nObject.toJSONString

        }).cache()

        //4、打印结果
        val df = spark.read.json(t1).select("Base_Info",
          "address",
          "administrative_region",
          "allItemCount",
          "brandFavCount",
          "brandName",
          "brandName_cn",
          "brandName_en",
          "brandSourceId",
          "brandValueId",
          "brand_isLaoZiHao",
          "brand_state",
          "brand_type",
          "brandallItemCount",
          "brandevaluates",
          "categoryName",
          "city",
          "city_grade",
          "city_origin",
          "crawl_date",
          "customCategoryId",
          "delivery_from",
          "district",
          "district_origin",
          "economic_division",
          "evaluates",
          "fans",
          "firstCategoryId",
          "fourthCategoryId",
          "goodRatePercentage",
          "goodUrl",
          "good_id",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "is_onsell",
          "is_oversea",
          "is_selfSupport",
          "name",
          "originCountryName",
          "originItem",
          "original_cost",
          "platformId",
          "platformName",
          "poor_counties",
          "priceText",
          "province",
          "regional_ID",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "score",
          "secondCategoryId",
          "sellCount",
          "sellCount_0601",
          "sellCount_0618",
          "shopId",
          "shopName",
          "shopType",
          "shopUrl",
          "subCategoryId",
          "subCategoryName",
          "taxFreeZone",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations")

//      df.show()
      df.repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/beibei/${month}")

      }
    }
  //3、提取qichezhijia
  def ES_OBS_3_qichezhijia(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {
        val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
        var t1 = value.map(x => {
          val nObject = JSON.parseObject(x)

          //1、将所有的字段转为string类型

          val has_stock  = nObject.getOrDefault("has_stock","-1").toString
          val good_id  = nObject.getOrDefault("good_id","-1").toString
          val fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
          val seriesId  = nObject.getOrDefault("seriesId","-1").toString
          val urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
          val economic_division  = nObject.getOrDefault("economic_division","-1").toString
          val province  = nObject.getOrDefault("province","-1").toString
          val poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
          val shopId  = nObject.getOrDefault("shopId","-1").toString
          val rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val longitude  = nObject.getOrDefault("longitude","-1").toString
          val goodRatePercentage  = nObject.getOrDefault("goodRatePercentage","-1").toString
          val specId  = nObject.getOrDefault("specId","-1").toString
          val brandName  = nObject.getOrDefault("brandName","-1").toString
          val images  = nObject.getOrDefault("images","-1").toString
          val goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
          val seriesName  = nObject.getOrDefault("seriesName","-1").toString
          val city_origin  = nObject.getOrDefault("city_origin","-1").toString
          val if_city  = nObject.getOrDefault("if_city","-1").toString
          val shopUrl  = nObject.getOrDefault("shopUrl","-1").toString
          val specName  = nObject.getOrDefault("specName","-1").toString
          val rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
          val district  = nObject.getOrDefault("district","-1").toString
          val sellCount  = nObject.getOrDefault("sellCount","-1").toString
          val district_origin  = nObject.getOrDefault("district_origin","-1").toString
          val shopType  = nObject.getOrDefault("shopType","-1").toString
          val the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
          val city  = nObject.getOrDefault("city","-1").toString
          val latitude  = nObject.getOrDefault("latitude","-1").toString
          val shopName  = nObject.getOrDefault("shopName","-1").toString
          val factory_name  = nObject.getOrDefault("factory_name","-1").toString
          val if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val title  = nObject.getOrDefault("title","-1").toString
          val city_grade  = nObject.getOrDefault("city_grade","-1").toString
          val firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
          val thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
          val salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
          val administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
          val platformName  = nObject.getOrDefault("platformName","-1").toString
          val priceText  = nObject.getOrDefault("priceText","-1").toString
          val if_district  = nObject.getOrDefault("if_district","-1").toString
          val address  = nObject.getOrDefault("address","-1").toString
          val the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val platformId  = nObject.getOrDefault("platformId","-1").toString
          val rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
          val timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
          val factory_id  = nObject.getOrDefault("factory_id","-1").toString
          val original_cost  = nObject.getOrDefault("original_cost","-1").toString
          val the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
          val the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
          val regional_ID  = nObject.getOrDefault("regional_ID","-1").toString


          nObject.put("has_stock",has_stock)
          nObject.put("good_id",good_id)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("seriesId",seriesId)
          nObject.put("urban_agglomerations",urban_agglomerations)
          nObject.put("economic_division",economic_division)
          nObject.put("province",province)
          nObject.put("poor_counties",poor_counties)
          nObject.put("shopId",shopId)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("longitude",longitude)
          nObject.put("goodRatePercentage",goodRatePercentage)
          nObject.put("specId",specId)
          nObject.put("brandName",brandName)
          nObject.put("images",images)
          nObject.put("goodUrl",goodUrl)
          nObject.put("seriesName",seriesName)
          nObject.put("city_origin",city_origin)
          nObject.put("if_city",if_city)
          nObject.put("shopUrl",shopUrl)
          nObject.put("specName",specName)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("district",district)
          nObject.put("sellCount",sellCount)
          nObject.put("district_origin",district_origin)
          nObject.put("shopType",shopType)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("city",city)
          nObject.put("latitude",latitude)
          nObject.put("shopName",shopName)
          nObject.put("factory_name",factory_name)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("title",title)
          nObject.put("city_grade",city_grade)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("salesAmount",salesAmount)
          nObject.put("administrative_region",administrative_region)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("if_district",if_district)
          nObject.put("address",address)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("platformId",platformId)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("factory_id",factory_id)
          nObject.put("original_cost",original_cost)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("brandValueId",brandValueId)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("regional_ID",regional_ID)


          //2、去掉某些字段
          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          nObject.toJSONString

        })

        //4、打印结果
        val df = spark.read.json(t1).select("address",
          "administrative_region",
          "brandName",
          "brandValueId",
          "city",
          "city_grade",
          "city_origin",
          "district",
          "district_origin",
          "economic_division",
          "factory_id",
          "factory_name",
          "firstCategoryId",
          "fourthCategoryId",
          "goodRatePercentage",
          "goodUrl",
          "good_id",
          "has_stock",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "latitude",
          "longitude",
          "original_cost",
          "platformId",
          "platformName",
          "poor_counties",
          "priceText",
          "province",
          "regional_ID",
          "rootCategoryId",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "secondCategoryId",
          "sellCount",
          "seriesId",
          "seriesName",
          "shopId",
          "shopName",
          "shopType",
          "shopUrl",
          "specId",
          "specName",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations")

          df.repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")
//      df.show()

      }
    }
  //4、提取babytree
  def ES_OBS_4_babytree(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

//      if(! (year.equals("2019") && month.equals("5"))){

        val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
        var t1 = value.map(x => {
          val nObject = JSON.parseObject(x)

          //1、将所有的字段转为string类型

          val  is_oversea  = nObject.getOrDefault("is_oversea","-1").toString
          val  fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
          val  good_id  = nObject.getOrDefault("good_id","-1").toString
          val  urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
          val  economic_division  = nObject.getOrDefault("economic_division","-1").toString
          val  province  = nObject.getOrDefault("province","-1").toString
          val  poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
          val  allItemCount  = nObject.getOrDefault("allItemCount","-1").toString
          val  brandSourceId  = nObject.getOrDefault("brandSourceId","-1").toString
          val  rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val  shopId  = nObject.getOrDefault("shopId","-1").toString
          val  sku  = nObject.getOrDefault("sku","-1").toString
          val  brand_type  = nObject.getOrDefault("brand_type","-1").toString
          val  longitude  = nObject.getOrDefault("longitude","-1").toString
          val  brandName  = nObject.getOrDefault("brandName","-1").toString
          val  images  = nObject.getOrDefault("images","-1").toString
          val  goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
          val  city_origin  = nObject.getOrDefault("city_origin","-1").toString
          val  if_city  = nObject.getOrDefault("if_city","-1").toString
          val  shopUrl  = nObject.getOrDefault("shopUrl","-1").toString
          val  brandName_en  = nObject.getOrDefault("brandName_en","-1").toString
          val  rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
          val  district  = nObject.getOrDefault("district","-1").toString
          val  taxFreeZone  = nObject.getOrDefault("taxFreeZone","-1").toString
          val  district_origin  = nObject.getOrDefault("district_origin","-1").toString
          val  name  = nObject.getOrDefault("name","-1").toString
          val  sellCount  = nObject.getOrDefault("sellCount","-1").toString
          val  shopType  = nObject.getOrDefault("shopType","-1").toString
          val  the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val  secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
          val  city  = nObject.getOrDefault("city","-1").toString
          val  Base_Info  = nObject.getOrDefault("Base_Info","-1").toString
          val  latitude  = nObject.getOrDefault("latitude","-1").toString
          val  brand_isLaoZiHao  = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val  shopName  = nObject.getOrDefault("shopName","-1").toString
          val  if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val  city_grade  = nObject.getOrDefault("city_grade","-1").toString
          val  title  = nObject.getOrDefault("title","-1").toString
          val  categoryName  = nObject.getOrDefault("categoryName","-1").toString
          val  firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
          val  thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
          val  salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
          val  administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
          val  platformName  = nObject.getOrDefault("platformName","-1").toString
          val  priceText  = nObject.getOrDefault("priceText","-1").toString
          val  if_district  = nObject.getOrDefault("if_district","-1").toString
          val  brand_state  = nObject.getOrDefault("brand_state","-1").toString
          val  address  = nObject.getOrDefault("address","-1").toString
          val  the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val  platformId  = nObject.getOrDefault("platformId","-1").toString
          val  rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
          val  timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
          val  original_cost  = nObject.getOrDefault("original_cost","-1").toString
          val  the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val  brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
          val  the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val  rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
          val  regional_ID  = nObject.getOrDefault("regional_ID","-1").toString
          val  brandName_cn  = nObject.getOrDefault("brandName_cn","-1").toString
          val  categoryId  = nObject.getOrDefault("categoryId","-1").toString


          nObject.put("is_oversea",is_oversea)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("good_id",good_id)
          nObject.put("urban_agglomerations",urban_agglomerations)
          nObject.put("economic_division",economic_division)
          nObject.put("province",province)
          nObject.put("poor_counties",poor_counties)
          nObject.put("allItemCount",allItemCount)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("shopId",shopId)
          nObject.put("sku",sku)
          nObject.put("brand_type",brand_type)
          nObject.put("longitude",longitude)
          nObject.put("brandName",brandName)
          nObject.put("images",images)
          nObject.put("goodUrl",goodUrl)
          nObject.put("city_origin",city_origin)
          nObject.put("if_city",if_city)
          nObject.put("shopUrl",shopUrl)
          nObject.put("brandName_en",brandName_en)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("district",district)
          nObject.put("taxFreeZone",taxFreeZone)
          nObject.put("district_origin",district_origin)
          nObject.put("name",name)
          nObject.put("sellCount",sellCount)
          nObject.put("shopType",shopType)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("city",city)
          nObject.put("Base_Info",Base_Info)
          nObject.put("latitude",latitude)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("shopName",shopName)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("city_grade",city_grade)
          nObject.put("title",title)
          nObject.put("categoryName",categoryName)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("salesAmount",salesAmount)
          nObject.put("administrative_region",administrative_region)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("if_district",if_district)
          nObject.put("brand_state",brand_state)
          nObject.put("address",address)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("platformId",platformId)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("original_cost",original_cost)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("brandValueId",brandValueId)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("regional_ID",regional_ID)
          nObject.put("brandName_cn",brandName_cn)
          nObject.put("categoryId",categoryId)

          //2、去掉某些字段
          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          nObject.toJSONString

        }).cache()

        //4、打印结果
        spark.read.json(t1).select("Base_Info",
          "address",
          "administrative_region",
          "allItemCount",
          "brandName",
          "brandName_cn",
          "brandName_en",
          "brandSourceId",
          "brandValueId",
          "brand_isLaoZiHao",
          "brand_state",
          "brand_type",
          "categoryId",
          "categoryName",
          "city",
          "city_grade",
          "city_origin",
          "district",
          "district_origin",
          "economic_division",
          "firstCategoryId",
          "fourthCategoryId",
          "goodUrl",
          "good_id",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "is_oversea",
          "latitude",
          "longitude",
          "name",
          "original_cost",
          "platformId",
          "platformName",
          "poor_counties",
          "priceText",
          "province",
          "regional_ID",
          "rootCategoryId",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "secondCategoryId",
          "sellCount",
          "shopId",
          "shopName",
          "shopType",
          "shopUrl",
          "sku",
          "taxFreeZone",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations")
          .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

//      }
    }
  }
  //5、提取mryx
  def ES_OBS_5_mryx(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

        val value: RDD[String] = spark.read.orc("s3a://dws-data/g_data/2021/2/mryx/").toJSON.rdd
//        val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
        var t1 = value.map(x => {
          val nObject = JSON.parseObject(x)

          //1、将所有的字段转为string类型

          val country  = nObject.getOrDefault("country","-1").toString
          val fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
          val good_id  = nObject.getOrDefault("good_id","-1").toString
          val urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
          val economic_division  = nObject.getOrDefault("economic_division","-1").toString
          val province  = nObject.getOrDefault("province","-1").toString
          val poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
          val brandSourceId  = nObject.getOrDefault("brandSourceId","-1").toString
          val rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val starts  = nObject.getOrDefault("starts","-1").toString
          val brand_type  = nObject.getOrDefault("brand_type","-1").toString
          val longitude  = nObject.getOrDefault("longitude","-1").toString
          val brandName  = nObject.getOrDefault("brandName","-1").toString
          val images  = nObject.getOrDefault("images","-1").toString
          val goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
          val city_origin  = nObject.getOrDefault("city_origin","-1").toString
          val if_city  = nObject.getOrDefault("if_city","-1").toString
          val subCategoryId  = nObject.getOrDefault("subCategoryId","-1").toString
          val min_delivery_price  = nObject.getOrDefault("min_delivery_price","-1").toString
          val brandName_en  = nObject.getOrDefault("brandName_en","-1").toString
          val rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
          val district  = nObject.getOrDefault("district","-1").toString
          val subtitle  = nObject.getOrDefault("subtitle","-1").toString
          val district_origin  = nObject.getOrDefault("district_origin","-1").toString
          val name  = nObject.getOrDefault("name","-1").toString
          val sellCount  = nObject.getOrDefault("sellCount","-1").toString
          val shopType  = nObject.getOrDefault("shopType","-1").toString
          val the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
          val city  = nObject.getOrDefault("city","-1").toString
          val Base_Info  = nObject.getOrDefault("Base_Info","-1").toString
          val latitude  = nObject.getOrDefault("latitude","-1").toString
          val brand_isLaoZiHao  = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val shopName  = nObject.getOrDefault("shopName","-1").toString
          val if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val city_grade  = nObject.getOrDefault("city_grade","-1").toString
          val title  = nObject.getOrDefault("title","-1").toString
          val categoryName  = nObject.getOrDefault("categoryName","-1").toString
          val firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
          val thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
          val subCategoryName  = nObject.getOrDefault("subCategoryName","-1").toString
          val order_lead_time  = nObject.getOrDefault("order_lead_time","-1").toString
          val salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
          val administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
          val platformName  = nObject.getOrDefault("platformName","-1").toString
          val priceText  = nObject.getOrDefault("priceText","-1").toString
          val if_district  = nObject.getOrDefault("if_district","-1").toString
          val brand_state  = nObject.getOrDefault("brand_state","-1").toString
          val coupon  = nObject.getOrDefault("coupon","-1").toString
          val the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val originProduction  = nObject.getOrDefault("originProduction","-1").toString
          val platformId  = nObject.getOrDefault("platformId","-1").toString
          val rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
          val timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
          val the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
          val the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
          val regional_ID  = nObject.getOrDefault("regional_ID","-1").toString
          val brandName_cn  = nObject.getOrDefault("brandName_cn","-1").toString
          val categoryId  = nObject.getOrDefault("categoryId","-1").toString


          nObject.put("country",country)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("good_id",good_id)
          nObject.put("urban_agglomerations",urban_agglomerations)
          nObject.put("economic_division",economic_division)
          nObject.put("province",province)
          nObject.put("poor_counties",poor_counties)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("starts",starts)
          nObject.put("brand_type",brand_type)
          nObject.put("longitude",longitude)
          nObject.put("brandName",brandName)
          nObject.put("images",images)
          nObject.put("goodUrl",goodUrl)
          nObject.put("city_origin",city_origin)
          nObject.put("if_city",if_city)
          nObject.put("subCategoryId",subCategoryId)
          nObject.put("min_delivery_price",min_delivery_price)
          nObject.put("brandName_en",brandName_en)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("district",district)
          nObject.put("subtitle",subtitle)
          nObject.put("district_origin",district_origin)
          nObject.put("name",name)
          nObject.put("sellCount",sellCount)
          nObject.put("shopType",shopType)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("city",city)
          nObject.put("Base_Info",Base_Info)
          nObject.put("latitude",latitude)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("shopName",shopName)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("city_grade",city_grade)
          nObject.put("title",title)
          nObject.put("categoryName",categoryName)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("subCategoryName",subCategoryName)
          nObject.put("order_lead_time",order_lead_time)
          nObject.put("salesAmount",salesAmount)
          nObject.put("administrative_region",administrative_region)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("if_district",if_district)
          nObject.put("brand_state",brand_state)
          nObject.put("coupon",coupon)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("originProduction",originProduction)
          nObject.put("platformId",platformId)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("brandValueId",brandValueId)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("regional_ID",regional_ID)
          nObject.put("brandName_cn",brandName_cn)
          nObject.put("categoryId",categoryId)


          //2、去掉某些字段
          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          nObject.toJSONString

        }).cache()

        //4、打印结果
        spark.read.json(t1).drop("shopId").registerTempTable("mryx_data")
      val addressdata = spark.read.option("header","true")
        .csv("s3a://o2o-dimension-table/address_table/address_table_2019/address_mapping_standard/2019_12/")
        .select("province","city","district","uniqueid").distinct().registerTempTable("addressdata")
      val data = spark.sql(
        """
          |select
          |t1.*,
          |concat('100',t2.uniqueid) as shopId
          |from
          |mryx_data t1
          |join
          |addressdata t2
          |on t1.province=t2.province and t1.city=t2.city and t1.district=t2.district
      """.stripMargin).dropDuplicates("good_id")
        .select("Base_Info",
          "administrative_region",
          "brandName",
          "brandName_cn",
          "brandName_en",
          "brandSourceId",
          "brandValueId",
          "brand_isLaoZiHao",
          "brand_state",
          "brand_type",
          "categoryId",
          "categoryName",
          "city",
          "city_grade",
          "city_origin",
          "country",
          "coupon",
          "district",
          "district_origin",
          "economic_division",
          "firstCategoryId",
          "fourthCategoryId",
          "goodUrl",
          "good_id",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "latitude",
          "longitude",
          "min_delivery_price",
          "name",
          "order_lead_time",
          "originProduction",
          "platformId",
          "platformName",
          "poor_counties",
          "priceText",
          "province",
          "regional_ID",
          "rootCategoryId",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "secondCategoryId",
          "sellCount",
          "shopId",
          "shopName",
          "shopType",
          "starts",
          "subCategoryId",
          "subCategoryName",
          "subtitle",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations")
        .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

      }
  }
  //6、提取hmxs
  def ES_OBS_6_hmxs(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

        val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
        var t1 = value.map(x => {
          val nObject = JSON.parseObject(x)
          val obj = new JSONObject()
          //1、将所有的字段转为string类型

          val companyName  = nObject.getOrDefault("companyName","-1").toString
          val fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
          val good_id  = nObject.getOrDefault("good_id","-1").toString
          val urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
          val economic_division  = nObject.getOrDefault("economic_division","-1").toString
          val province  = nObject.getOrDefault("province","-1").toString
          val poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
          val tbSellCount  = nObject.getOrDefault("tbSellCount","-1").toString
          val rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val shopId  = nObject.getOrDefault("shopId","-1").toString
          val brand_type  = nObject.getOrDefault("brand_type","-1").toString
          val goodRatePercentage  = nObject.getOrDefault("goodRatePercentage","-1").toString
          val brandName  = nObject.getOrDefault("brandName","-1").toString
          val images  = nObject.getOrDefault("images","-1").toString
          val goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
          val city_origin  = nObject.getOrDefault("city_origin","-1").toString
          val if_city  = nObject.getOrDefault("if_city","-1").toString
          val crawl_date  = nObject.getOrDefault("crawl_date","-1").toString
          val subCategoryId  = nObject.getOrDefault("subCategoryId","-1").toString
          val commentCount  = nObject.getOrDefault("commentCount","-1").toString
          val companyId  = nObject.getOrDefault("companyId","-1").toString
          val brandName_en  = nObject.getOrDefault("brandName_en","-1").toString
          val rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
          val district  = nObject.getOrDefault("district","-1").toString
          val taxFreeZone  = nObject.getOrDefault("taxFreeZone","-1").toString
          val district_origin  = nObject.getOrDefault("district_origin","-1").toString
          val sellCount  = nObject.getOrDefault("sellCount","-1").toString
          val name  = nObject.getOrDefault("name","-1").toString
          val shopType  = nObject.getOrDefault("shopType","-1").toString
          val the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
          val city  = nObject.getOrDefault("city","-1").toString
          val Base_Info  = nObject.getOrDefault("Base_Info","-1").toString
          val brand_isLaoZiHao  = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val shopName  = nObject.getOrDefault("shopName","-1").toString
          val if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val city_grade  = nObject.getOrDefault("city_grade","-1").toString
          val title  = nObject.getOrDefault("title","-1").toString
          val categoryName  = nObject.getOrDefault("categoryName","-1").toString
          val firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
          val thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
          val subCategoryName  = nObject.getOrDefault("subCategoryName","-1").toString
          val salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
          val administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
          val platformName  = nObject.getOrDefault("platformName","-1").toString
          val priceText  = nObject.getOrDefault("priceText","-1").toString
          val if_district  = nObject.getOrDefault("if_district","-1").toString
          val brand_state  = nObject.getOrDefault("brand_state","-1").toString
          val address  = nObject.getOrDefault("address","-1").toString
          val tbCategoryId  = nObject.getOrDefault("tbCategoryId","-1").toString
          val the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val platformId  = nObject.getOrDefault("platformId","-1").toString
          val rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
          val tbBrandValueId  = nObject.getOrDefault("tbBrandValueId","-1").toString
          val timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
          val original_cost  = nObject.getOrDefault("original_cost","-1").toString
          val the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
          val originItem  = nObject.getOrDefault("originItem","-1").toString
          val the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val originCountryName  = nObject.getOrDefault("originCountryName","-1").toString
          val rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
          val regional_ID  = nObject.getOrDefault("regional_ID","-1").toString
          val brandName_cn  = nObject.getOrDefault("brandName_cn","-1").toString
          val categoryId  = nObject.getOrDefault("categoryId","-1").toString

          obj.put("companyName",companyName)
          obj.put("fourthCategoryId",fourthCategoryId)
          obj.put("good_id",good_id)
          obj.put("urban_agglomerations",urban_agglomerations)
          obj.put("economic_division",economic_division)
          obj.put("province",province)
          obj.put("poor_counties",poor_counties)
          obj.put("tbSellCount",tbSellCount)
          obj.put("rural_demonstration_counties",rural_demonstration_counties)
          obj.put("shopId",shopId)
          obj.put("brand_type",brand_type)
          obj.put("goodRatePercentage",goodRatePercentage)
          obj.put("brandName",brandName)
          obj.put("images",images)
          obj.put("goodUrl",goodUrl)
          obj.put("city_origin",city_origin)
          obj.put("if_city",if_city)
          obj.put("crawl_date",crawl_date)
          obj.put("subCategoryId",subCategoryId)
          obj.put("commentCount",commentCount)
          obj.put("companyId",companyId)
          obj.put("brandName_en",brandName_en)
          obj.put("rural_ecommerce",rural_ecommerce)
          obj.put("district",district)
          obj.put("taxFreeZone",taxFreeZone)
          obj.put("district_origin",district_origin)
          obj.put("sellCount",sellCount)
          obj.put("name",name)
          obj.put("shopType",shopType)
          obj.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          obj.put("secondCategoryId",secondCategoryId)
          obj.put("city",city)
          obj.put("Base_Info",Base_Info)
          obj.put("brand_isLaoZiHao",brand_isLaoZiHao)
          obj.put("shopName",shopName)
          obj.put("if_state_level_new_areas",if_state_level_new_areas)
          obj.put("city_grade",city_grade)
          obj.put("title",title)
          obj.put("categoryName",categoryName)
          obj.put("firstCategoryId",firstCategoryId)
          obj.put("thirdCategoryId",thirdCategoryId)
          obj.put("subCategoryName",subCategoryName)
          obj.put("salesAmount",salesAmount)
          obj.put("administrative_region",administrative_region)
          obj.put("platformName",platformName)
          obj.put("priceText",priceText)
          obj.put("if_district",if_district)
          obj.put("brand_state",brand_state)
          obj.put("address",address)
          obj.put("tbCategoryId",tbCategoryId)
          obj.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          obj.put("platformId",platformId)
          obj.put("rootCategoryId",rootCategoryId)
          obj.put("tbBrandValueId",tbBrandValueId)
          obj.put("timeStamp",timeStamp)
          obj.put("original_cost",original_cost)
          obj.put("the_belt_and_road_city",the_belt_and_road_city)
          obj.put("brandValueId",brandValueId)
          obj.put("originItem",originItem)
          obj.put("the_belt_and_road_province",the_belt_and_road_province)
          obj.put("originCountryName",originCountryName)
          obj.put("rootCategoryName",rootCategoryName)
          obj.put("regional_ID",regional_ID)
          obj.put("brandName_cn",brandName_cn)
          obj.put("categoryId",categoryId)


          //2、去掉某些字段
          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          obj.toJSONString

        }).cache()

        //4、打印结果
        //      ssc.read.json(t1).show()
        spark.read.json(t1).select("Base_Info",
          "address",
          "administrative_region",
          "brandName",
          "brandName_cn",
          "brandName_en",
          "brandValueId",
          "brand_isLaoZiHao",
          "brand_state",
          "brand_type",
          "categoryId",
          "categoryName",
          "city",
          "city_grade",
          "city_origin",
          "commentCount",
          "companyId",
          "companyName",
          "crawl_date",
          "district",
          "district_origin",
          "economic_division",
          "firstCategoryId",
          "fourthCategoryId",
          "goodRatePercentage",
          "goodUrl",
          "good_id",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "name",
          "originCountryName",
          "originItem",
          "original_cost",
          "platformId",
          "platformName",
          "poor_counties",
          "priceText",
          "province",
          "regional_ID",
          "rootCategoryId",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "secondCategoryId",
          "sellCount",
          "shopId",
          "shopName",
          "shopType",
          "subCategoryId",
          "subCategoryName",
          "taxFreeZone",
          "tbBrandValueId",
          "tbCategoryId",
          "tbSellCount",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations")
          .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

      }
  }
  //7、提取shunfeng
  def ES_OBS_7_shunfeng(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

//      if(!(year.equals("2019") && month.equals("5"))){

        val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
        var t1 = value.map(x => {
          val nObject = JSON.parseObject(x)

          //1、将所有的字段转为string类型

          val is_oversea  = nObject.getOrDefault("is_oversea","-1").toString
          val fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
          val good_id  = nObject.getOrDefault("good_id","-1").toString
          val urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
          val economic_division  = nObject.getOrDefault("economic_division","-1").toString
          val province  = nObject.getOrDefault("province","-1").toString
          val poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
          val brandSourceId  = nObject.getOrDefault("brandSourceId","-1").toString
          val rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val brand_type  = nObject.getOrDefault("brand_type","-1").toString
          val goodRatePercentage  = nObject.getOrDefault("goodRatePercentage","-1").toString
          val brandName  = nObject.getOrDefault("brandName","-1").toString
          val images  = nObject.getOrDefault("images","-1").toString
          val goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
          val city_origin  = nObject.getOrDefault("city_origin","-1").toString
          val if_city  = nObject.getOrDefault("if_city","-1").toString
          val subCategoryId  = nObject.getOrDefault("subCategoryId","-1").toString
          val brandName_en  = nObject.getOrDefault("brandName_en","-1").toString
          val rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
          val district  = nObject.getOrDefault("district","-1").toString
          val delivery_from  = nObject.getOrDefault("delivery_from","-1").toString
          val district_origin  = nObject.getOrDefault("district_origin","-1").toString
          val name  = nObject.getOrDefault("name","-1").toString
          val sellCount  = nObject.getOrDefault("sellCount","-1").toString
          val shopType  = nObject.getOrDefault("shopType","-1").toString
          val the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val vip_price  = nObject.getOrDefault("vip_price","-1").toString
          val secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
          val city  = nObject.getOrDefault("city","-1").toString
          val Base_Info  = nObject.getOrDefault("Base_Info","-1").toString
          val brand_isLaoZiHao  = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val shopName  = nObject.getOrDefault("shopName","-1").toString
          val if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val city_grade  = nObject.getOrDefault("city_grade","-1").toString
          val title  = nObject.getOrDefault("title","-1").toString
          val categoryName  = nObject.getOrDefault("categoryName","-1").toString
          val firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
          val thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
          val subCategoryName  = nObject.getOrDefault("subCategoryName","-1").toString
          val goodCount  = nObject.getOrDefault("goodCount","-1").toString
          val salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
          val administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
          val platformName  = nObject.getOrDefault("platformName","-1").toString
          val priceText  = nObject.getOrDefault("priceText","-1").toString
          val if_district  = nObject.getOrDefault("if_district","-1").toString
          val poorCount  = nObject.getOrDefault("poorCount","-1").toString
          val brand_state  = nObject.getOrDefault("brand_state","-1").toString
          val address  = nObject.getOrDefault("address","-1").toString
          val the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val originProduction  = nObject.getOrDefault("originProduction","-1").toString
          val platformId  = nObject.getOrDefault("platformId","-1").toString
          val rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
          val timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
          val original_cost  = nObject.getOrDefault("original_cost","-1").toString
          val generalCount  = nObject.getOrDefault("generalCount","-1").toString
          val the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
          val originItem  = nObject.getOrDefault("originItem","-1").toString
          val the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val originCountryName  = nObject.getOrDefault("originCountryName","-1").toString
          val is_selfSupport  = nObject.getOrDefault("is_selfSupport","-1").toString
          val rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
          val agric_products  = nObject.getOrDefault("agric_products","-1").toString
          val regional_ID  = nObject.getOrDefault("regional_ID","-1").toString
          val platformName_spelling  = nObject.getOrDefault("platformName_spelling","-1").toString
          val brandName_cn  = nObject.getOrDefault("brandName_cn","-1").toString
          val categoryId  = nObject.getOrDefault("categoryId","-1").toString
          val customCategoryId  = nObject.getOrDefault("customCategoryId","-1").toString

          nObject.put("is_oversea",is_oversea)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("good_id",good_id)
          nObject.put("urban_agglomerations",urban_agglomerations)
          nObject.put("economic_division",economic_division)
          nObject.put("province",province)
          nObject.put("poor_counties",poor_counties)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("brand_type",brand_type)
          nObject.put("goodRatePercentage",goodRatePercentage)
          nObject.put("brandName",brandName)
          nObject.put("images",images)
          nObject.put("goodUrl",goodUrl)
          nObject.put("city_origin",city_origin)
          nObject.put("if_city",if_city)
          nObject.put("subCategoryId",subCategoryId)
          nObject.put("brandName_en",brandName_en)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("district",district)
          nObject.put("delivery_from",delivery_from)
          nObject.put("district_origin",district_origin)
          nObject.put("name",name)
          nObject.put("sellCount",sellCount)
          nObject.put("shopType",shopType)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("vip_price",vip_price)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("city",city)
          nObject.put("Base_Info",Base_Info)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("shopName",shopName)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("city_grade",city_grade)
          nObject.put("title",title)
          nObject.put("categoryName",categoryName)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("subCategoryName",subCategoryName)
          nObject.put("goodCount",goodCount)
          nObject.put("salesAmount",salesAmount)
          nObject.put("administrative_region",administrative_region)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("if_district",if_district)
          nObject.put("poorCount",poorCount)
          nObject.put("brand_state",brand_state)
          nObject.put("address",address)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("originProduction",originProduction)
          nObject.put("platformId",platformId)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("original_cost",original_cost)
          nObject.put("generalCount",generalCount)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("brandValueId",brandValueId)
          nObject.put("originItem",originItem)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("originCountryName",originCountryName)
          nObject.put("is_selfSupport",is_selfSupport)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("agric_products",agric_products)
          nObject.put("regional_ID",regional_ID)
          nObject.put("platformName_spelling",platformName_spelling)
          nObject.put("brandName_cn",brandName_cn)
          nObject.put("categoryId",categoryId)
          nObject.put("customCategoryId",customCategoryId)

          //2、去掉某些字段
          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          nObject.toJSONString

        }).cache()

        //4、打印结果
        //      ssc.read.json(t1).show()
        spark.read.json(t1).select("Base_Info",
          "address",
          "administrative_region",
          "agric_products",
          "brandName",
          "brandName_cn",
          "brandName_en",
          "brandSourceId",
          "brandValueId",
          "brand_isLaoZiHao",
          "brand_state",
          "brand_type",
          "categoryId",
          "categoryName",
          "city",
          "city_grade",
          "city_origin",
          "customCategoryId",
          "delivery_from",
          "district",
          "district_origin",
          "economic_division",
          "firstCategoryId",
          "fourthCategoryId",
          "generalCount",
          "goodCount",
          "goodRatePercentage",
          "goodUrl",
          "good_id",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "is_oversea",
          "is_selfSupport",
          "name",
          "originCountryName",
          "originItem",
          "originProduction",
          "original_cost",
          "platformId",
          "platformName",
          "platformName_spelling",
          "poorCount",
          "poor_counties",
          "priceText",
          "province",
          "regional_ID",
          "rootCategoryId",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "secondCategoryId",
          "sellCount",
          "shopName",
          "shopType",
          "subCategoryId",
          "subCategoryName",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations",
          "vip_price")
          .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

//      }
    }
  }
  //8、提取miya
  def ES_OBS_8_miya(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {
    val sc = spark.sparkContext
    for (month <- monthArr) {

        val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
        var t1 = value.map(x => {
          val nObject = JSON.parseObject(x)

          //1、将所有的字段转为string类型
          val is_oversea  = nObject.getOrDefault("is_oversea","-1").toString
          val fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
          val good_id  = nObject.getOrDefault("good_id","-1").toString
          val urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
          val economic_division  = nObject.getOrDefault("economic_division","-1").toString
          val province  = nObject.getOrDefault("province","-1").toString
          val poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
          val brandSourceId  = nObject.getOrDefault("brandSourceIdb","-1").toString
          val rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val shopId  = nObject.getOrDefault("shopId","-1").toString
          val brand_type  = nObject.getOrDefault("brand_type","-1").toString
          val created_time  = nObject.getOrDefault("created_time","-1").toString
          val brandName  = nObject.getOrDefault("brandName","-1").toString
          val images  = nObject.getOrDefault("images","-1").toString
          val goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
          val city_origin  = nObject.getOrDefault("city_origin","-1").toString
          val if_city  = nObject.getOrDefault("if_city","-1").toString
          val subCategoryId  = nObject.getOrDefault("subCategoryId","-1").toString

          val commentCount  = nObject.getOrDefault("commentCount","-1").toString

          val is_onsell  = nObject.getOrDefault("is_onsell","-1").toString
          val brandName_en  = nObject.getOrDefault("brandName_enn","-1").toString
          val rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
          val district  = nObject.getOrDefault("district","-1").toString
          val taxFreeZone  = nObject.getOrDefault("taxFreeZone","-1").toString
          val district_origin  = nObject.getOrDefault("district_origin","-1").toString
          val name  = nObject.getOrDefault("name","-1").toString
          val sellCount  = nObject.getOrDefault("sellCount","-1").toString
          val shopType  = nObject.getOrDefault("shopType","-1").toString
          val the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
          val penghuaName  = nObject.getOrDefault("penghuaName","-1").toString
          val city  = nObject.getOrDefault("city","-1").toString
          val Base_Info  = nObject.getOrDefault("Base_Info","-1").toString
          val brand_isLaoZiHao  = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val shopName  = nObject.getOrDefault("shopName","-1").toString
          val if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val city_grade  = nObject.getOrDefault("city_grade","-1").toString
          val title  = nObject.getOrDefault("title","-1").toString
          val categoryName  = nObject.getOrDefault("categoryName","-1").toString
          val firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
          val thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
          val subCategoryName  = nObject.getOrDefault("subCategoryName","-1").toString
          val commentator  = nObject.getOrDefault("commentator","-1").toString
          val salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
          val administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
          val platformName  = nObject.getOrDefault("platformName","-1").toString
          val priceText  = nObject.getOrDefault("priceText","-1").toString
          val if_district  = nObject.getOrDefault("if_district","-1").toString
          val brand_state  = nObject.getOrDefault("brand_state","-1").toString
          val address  = nObject.getOrDefault("address","-1").toString
          val the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val originProduction  = nObject.getOrDefault("originProduction","-1").toString
          val platformId  = nObject.getOrDefault("platformId","-1").toString
          val rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
          val timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
          val original_cost  = nObject.getOrDefault("original_cost","-1").toString
          val the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
          val the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val originCountryName  = nObject.getOrDefault("originCountryName","-1").toString
          val is_selfSupport  = nObject.getOrDefault("is_selfSupport","-1").toString
          val rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
          val regional_ID  = nObject.getOrDefault("regional_ID","-1").toString
          val brandName_cn  = nObject.getOrDefault("brandName_cn","-1").toString

          val group_num  = nObject.getOrDefault("group_num","-1").toString
          val group_price  = nObject.getOrDefault("group_price","-1").toString
          val group_sellCount  = nObject.getOrDefault("group_sellCount","-1").toString
          val group_soldNum  = nObject.getOrDefault("group_soldNum","-1").toString
          val products  = nObject.getOrDefault("products","-1").toString


          nObject.put("is_oversea",is_oversea)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("good_id",good_id)
          nObject.put("urban_agglomerations",urban_agglomerations)
          nObject.put("economic_division",economic_division)
          nObject.put("province",province)
          nObject.put("poor_counties",poor_counties)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("shopId",shopId)
          nObject.put("brand_type",brand_type)
          nObject.put("created_time",created_time)
          nObject.put("brandName",brandName)
          nObject.put("images",images)
          nObject.put("goodUrl",goodUrl)
          nObject.put("city_origin",city_origin)
          nObject.put("if_city",if_city)
          nObject.put("subCategoryId",subCategoryId)
          nObject.put("commentCount",commentCount)
          nObject.put("is_onsell",is_onsell)
          nObject.put("brandName_en",brandName_en)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("district",district)
          nObject.put("taxFreeZone",taxFreeZone)
          nObject.put("district_origin",district_origin)
          nObject.put("name",name)
          nObject.put("sellCount",sellCount)
          nObject.put("shopType",shopType)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("penghuaName",penghuaName)
          nObject.put("city",city)
          nObject.put("Base_Info",Base_Info)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("shopName",shopName)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("city_grade",city_grade)
          nObject.put("title",title)
          nObject.put("categoryName",categoryName)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("subCategoryName",subCategoryName)
          nObject.put("commentator",commentator)
          nObject.put("salesAmount",salesAmount)
          nObject.put("administrative_region",administrative_region)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("if_district",if_district)
          nObject.put("brand_state",brand_state)
          nObject.put("address",address)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("originProduction",originProduction)
          nObject.put("platformId",platformId)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("original_cost",original_cost)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("brandValueId",brandValueId)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("originCountryName",originCountryName)
          nObject.put("is_selfSupport",is_selfSupport)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("regional_ID",regional_ID)
          nObject.put("brandName_cn",brandName_cn)

          nObject.put("group_num",group_num)
          nObject.put("group_price",group_price)
          nObject.put("group_sellCount",group_sellCount)
          nObject.put("group_soldNum",group_soldNum)
          nObject.put("products",products)

          //2、去掉某些字段
          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          nObject.toJSONString

        })

        //4、打印结果
        spark.read.json(t1)
          .drop("city").withColumn("city",lit("北京市"))
          .drop("shopId")
          .registerTempTable("mryx_data")

      val addressdata = spark.read.option("header","true")
        .csv("s3a://o2o-dimension-table/address_table/address_table_2019/address_mapping_standard/2019_12/")
        .select("province","city","district","uniqueid").distinct().registerTempTable("addressdata")
      val data = spark.sql(
        """
          |select
          |t1.*,
          |concat('100',t2.uniqueid) as shopId
          |from
          |mryx_data t1
          |join
          |addressdata t2
          |on t1.province=t2.province and t1.city=t2.city and t1.district=t2.district
      """.stripMargin).dropDuplicates("good_id")
        .select("Base_Info",
          "address",
          "administrative_region",
          "brandName",
          "brandName_cn",
          "brandName_en",
          "brandSourceId",
          "brandValueId",
          "brand_isLaoZiHao",
          "brand_state",
          "brand_type",
          "categoryName",
          "city",
          "city_grade",
          "city_origin",
          "commentCount",
          "commentator",
          "created_time",
          "district",
          "district_origin",
          "economic_division",
          "firstCategoryId",
          "fourthCategoryId",
          "goodUrl",
          "good_id",
          "group_num",
          "group_price",
          "group_sellCount",
          "group_soldNum",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "is_onsell",
          "is_oversea",
          "is_selfSupport",
          "name",
          "originCountryName",
          "originProduction",
          "original_cost",
          "penghuaName",
          "platformId",
          "platformName",
          "poor_counties",
          "priceText",
          "products",
          "province",
          "regional_ID",
          "rootCategoryId",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "secondCategoryId",
          "sellCount",
          "shopId",
          "shopName",
          "shopType",
          "subCategoryId",
          "subCategoryName",
          "taxFreeZone",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations")
        .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

      }
  }
  //9、提取kaola
  def ES_OBS_9_kaola(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

        val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
        var t1 = value.map(x => {
          val nObject = JSON.parseObject(x)

          //1、将所有的字段转为string类型

          val is_oversea  = nObject.getOrDefault("is_oversea","-1").toString
          val fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
          val good_id  = nObject.getOrDefault("good_id","-1").toString
          val urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
          val economic_division  = nObject.getOrDefault("economic_division","-1").toString
          val province  = nObject.getOrDefault("province","-1").toString
          val poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
          val sellCountNeed2  = nObject.getOrDefault("sellCountNeed2","-1").toString
          val goodsTaxAmount  = nObject.getOrDefault("goodsTaxAmount","-1").toString
          val brandSourceId  = nObject.getOrDefault("brandSourceId","-1").toString
          val rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
          val shopId  = nObject.getOrDefault("shopId","-1").toString
          val brand_type  = nObject.getOrDefault("brand_type","-1").toString
          val goodRatePercentage  = nObject.getOrDefault("goodRatePercentage","-1").toString
          val brandName  = nObject.getOrDefault("brandName","-1").toString
          val images  = nObject.getOrDefault("images","-1").toString
          val goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
          val city_origin  = nObject.getOrDefault("city_origin","-1").toString
          val if_city  = nObject.getOrDefault("if_city","-1").toString
          val subCategoryId  = nObject.getOrDefault("subCategoryId","-1").toString
          val shopUrl  = nObject.getOrDefault("shopUrl","-1").toString
          val quantity_jian_lt0  = nObject.getOrDefault("quantity_jian_lt0","-1").toString
          val detailCategoryName  = nObject.getOrDefault("detailCategoryName","-1").toString
          val brandName_en  = nObject.getOrDefault("brandName_en","-1").toString
          val rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
          val district  = nObject.getOrDefault("district","-1").toString
          val delivery_from  = nObject.getOrDefault("delivery_from","-1").toString
          val district_origin  = nObject.getOrDefault("district_origin","-1").toString
          val name  = nObject.getOrDefault("name","-1").toString
          val sellCount  = nObject.getOrDefault("sellCount","-1").toString
          val shopType  = nObject.getOrDefault("shopType","-1").toString
          val the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
          val secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
          val is_goldSeller  = nObject.getOrDefault("is_goldSeller","-1").toString
          val city  = nObject.getOrDefault("city","-1").toString
          val Base_Info  = nObject.getOrDefault("Base_Info","-1").toString
          val detailCategoryId  = nObject.getOrDefault("detailCategoryId","-1").toString
          val brand_isLaoZiHao  = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
          val shopName  = nObject.getOrDefault("shopName","-1").toString
          val if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
          val city_grade  = nObject.getOrDefault("city_grade","-1").toString
          val shangGt1  = nObject.getOrDefault("shangGt1","-1").toString
          val title  = nObject.getOrDefault("title","-1").toString
          val categoryName  = nObject.getOrDefault("categoryName","-1").toString
          val firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
          val thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
          val subCategoryName  = nObject.getOrDefault("subCategoryName","-1").toString
          val avg  = nObject.getOrDefault("avg","-1").toString
          val goodCount  = nObject.getOrDefault("goodCount","-1").toString
          val rate  = nObject.getOrDefault("rate","-1").toString
          val salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
          val administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
          val platformName  = nObject.getOrDefault("platformName","-1").toString
          val priceText  = nObject.getOrDefault("priceText","-1").toString
          val if_district  = nObject.getOrDefault("if_district","-1").toString
          val brand_state  = nObject.getOrDefault("brand_state","-1").toString
          val address  = nObject.getOrDefault("address","-1").toString
          val the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
          val platformId  = nObject.getOrDefault("platformId","-1").toString
          val rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
          val timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
          val original_cost  = nObject.getOrDefault("original_cost","-1").toString
          val standT  = nObject.getOrDefault("standT","-1").toString
          val the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
          val brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
          val originItem  = nObject.getOrDefault("originItem","-1").toString
          val the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
          val originCountryName  = nObject.getOrDefault("originCountryName","-1").toString
          val is_selfSupport  = nObject.getOrDefault("is_selfSupport","-1").toString
          val sellCountNeed  = nObject.getOrDefault("sellCountNeed","-1").toString
          val rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
          val regional_ID  = nObject.getOrDefault("regional_ID","-1").toString
          val brandName_cn  = nObject.getOrDefault("brandName_cn","-1").toString
          val categoryId  = nObject.getOrDefault("categoryId","-1").toString

          val model  = nObject.getOrDefault("model","-1").toString
          val color  = nObject.getOrDefault("color","-1").toString
          val ram  = nObject.getOrDefault("ram","-1").toString
          val rom  = nObject.getOrDefault("rom","-1").toString
          val series = nObject.getOrDefault("series","-1").toString

          nObject.put("is_oversea",is_oversea)
          nObject.put("fourthCategoryId",fourthCategoryId)
          nObject.put("good_id",good_id)
          nObject.put("urban_agglomerations",urban_agglomerations)
          nObject.put("economic_division",economic_division)
          nObject.put("province",province)
          nObject.put("poor_counties",poor_counties)
          nObject.put("sellCountNeed2",sellCountNeed2)
          nObject.put("goodsTaxAmount",goodsTaxAmount)
          nObject.put("brandSourceId",brandSourceId)
          nObject.put("rural_demonstration_counties",rural_demonstration_counties)
          nObject.put("shopId",shopId)
          nObject.put("brand_type",brand_type)
          nObject.put("goodRatePercentage",goodRatePercentage)
          nObject.put("brandName",brandName)
          nObject.put("images",images)
          nObject.put("goodUrl",goodUrl)
          nObject.put("city_origin",city_origin)
          nObject.put("if_city",if_city)
          nObject.put("subCategoryId",subCategoryId)
          nObject.put("shopUrl",shopUrl)
          nObject.put("quantity_jian_lt0",quantity_jian_lt0)
          nObject.put("detailCategoryName",detailCategoryName)
          nObject.put("brandName_en",brandName_en)
          nObject.put("rural_ecommerce",rural_ecommerce)
          nObject.put("district",district)
          nObject.put("delivery_from",delivery_from)
          nObject.put("district_origin",district_origin)
          nObject.put("name",name)
          nObject.put("sellCount",sellCount)
          nObject.put("shopType",shopType)
          nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
          nObject.put("secondCategoryId",secondCategoryId)
          nObject.put("is_goldSeller",is_goldSeller)
          nObject.put("city",city)
          nObject.put("Base_Info",Base_Info)
          nObject.put("detailCategoryId",detailCategoryId)
          nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
          nObject.put("shopName",shopName)
          nObject.put("if_state_level_new_areas",if_state_level_new_areas)
          nObject.put("city_grade",city_grade)
          nObject.put("shangGt1",shangGt1)
          nObject.put("title",title)
          nObject.put("categoryName",categoryName)
          nObject.put("firstCategoryId",firstCategoryId)
          nObject.put("thirdCategoryId",thirdCategoryId)
          nObject.put("subCategoryName",subCategoryName)
          nObject.put("avg",avg)
          nObject.put("goodCount",goodCount)
          nObject.put("rate",rate)
          nObject.put("salesAmount",salesAmount)
          nObject.put("administrative_region",administrative_region)
          nObject.put("platformName",platformName)
          nObject.put("priceText",priceText)
          nObject.put("if_district",if_district)
          nObject.put("brand_state",brand_state)
          nObject.put("address",address)
          nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
          nObject.put("platformId",platformId)
          nObject.put("rootCategoryId",rootCategoryId)
          nObject.put("timeStamp",timeStamp)
          nObject.put("original_cost",original_cost)
          nObject.put("standT",standT)
          nObject.put("the_belt_and_road_city",the_belt_and_road_city)
          nObject.put("brandValueId",brandValueId)
          nObject.put("originItem",originItem)
          nObject.put("the_belt_and_road_province",the_belt_and_road_province)
          nObject.put("originCountryName",originCountryName)
          nObject.put("is_selfSupport",is_selfSupport)
          nObject.put("sellCountNeed",sellCountNeed)
          nObject.put("rootCategoryName",rootCategoryName)
          nObject.put("regional_ID",regional_ID)
          nObject.put("brandName_cn",brandName_cn)
          nObject.put("categoryId",categoryId)

          nObject.put("model",model)
          nObject.put("color",color)
          nObject.put("ram",ram)
          nObject.put("rom",rom)
          nObject.put("series",series)

          //2、去掉某些字段
          nObject.remove("add_to_field")
          nObject.remove("function")
          nObject.remove("shangGt1")

          nObject.toJSONString

        })

        //4、打印结果
        spark.read.json(t1).select("Base_Info",
          "address",
          "administrative_region",
          "avg",
          "brandName",
          "brandName_cn",
          "brandName_en",
          "brandSourceId",
          "brandValueId",
          "brand_isLaoZiHao",
          "brand_state",
          "brand_type",
          "categoryId",
          "categoryName",
          "city",
          "city_grade",
          "city_origin",
          "color",
          "delivery_from",
          "detailCategoryId",
          "detailCategoryName",
          "district",
          "district_origin",
          "economic_division",
          "firstCategoryId",
          "fourthCategoryId",
          "goodCount",
          "goodRatePercentage",
          "goodUrl",
          "good_id",
          "goodsTaxAmount",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "is_goldSeller",
          "is_oversea",
          "is_selfSupport",
          "model",
          "name",
          "originCountryName",
          "originItem",
          "original_cost",
          "platformId",
          "platformName",
          "poor_counties",
          "priceText",
          "province",
          "quantity_jian_lt0",
          "ram",
          "rate",
          "regional_ID",
          "rom",
          "rootCategoryId",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "secondCategoryId",
          "sellCount",
          "sellCountNeed",
          "sellCountNeed2",
          "series",
          "shopId",
          "shopName",
          "shopType",
          "shopUrl",
          "standT",
          "subCategoryId",
          "subCategoryName",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations")
          .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

      }

  }
  //10、提取secoo
  def ES_OBS_10_secoo(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {
      val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
      var t1 = value.map(x => {
        val nObject = JSON.parseObject(x)

        //1、将所有的字段转为string类型

        val is_oversea = nObject.getOrDefault("is_oversea", "-1").toString
        val fourthCategoryId = nObject.getOrDefault("fourthCategoryId", "-1").toString
        val good_id = nObject.getOrDefault("good_id", "-1").toString
        val urban_agglomerations = nObject.getOrDefault("urban_agglomerations", "-1").toString
        val economic_division = nObject.getOrDefault("economic_division", "-1").toString
        val province = nObject.getOrDefault("province", "-1").toString
        val poor_counties = nObject.getOrDefault("poor_counties", "-1").toString
        val brandSourceId = nObject.getOrDefault("brandSourceId", "-1").toString
        val rural_demonstration_counties = nObject.getOrDefault("rural_demonstration_counties", "-1").toString
        val shopId = nObject.getOrDefault("shopId", "-1").toString
        val brand_type = nObject.getOrDefault("brand_type", "-1").toString
        val brandName = nObject.getOrDefault("brandName", "-1").toString
        val images = nObject.getOrDefault("images", "-1").toString
        val good_code = nObject.getOrDefault("good_code", "-1").toString
        val goodUrl = nObject.getOrDefault("goodUrl", "-1").toString
        val city_origin = nObject.getOrDefault("city_origin", "-1").toString
        val if_city = nObject.getOrDefault("if_city", "-1").toString
        val crawl_date = nObject.getOrDefault("crawl_date", "-1").toString
        val subCategoryId = nObject.getOrDefault("subCategoryId", "-1").toString
        val commentCount = nObject.getOrDefault("commentCount", "-1").toString
        val is_onsell = nObject.getOrDefault("is_onsell", "-1").toString
        val brandName_en = nObject.getOrDefault("brandName_en", "-1").toString
        val rural_ecommerce = nObject.getOrDefault("rural_ecommerce", "-1").toString
        val district = nObject.getOrDefault("district", "-1").toString
        val delivery_from = nObject.getOrDefault("delivery_from", "-1").toString
        val district_origin = nObject.getOrDefault("district_origin", "-1").toString
        val name = nObject.getOrDefault("name", "-1").toString
        val sellCount = nObject.getOrDefault("sellCount", "-1").toString
        val shopType = nObject.getOrDefault("shopType", "-1").toString
        val the_yangtze_river_economic_zone_province = nObject.getOrDefault("the_yangtze_river_economic_zone_province", "-1").toString
        val secondCategoryId = nObject.getOrDefault("secondCategoryId", "-1").toString
        val city = nObject.getOrDefault("city", "-1").toString
        val Base_Info = nObject.getOrDefault("Base_Info", "-1").toString
        val brand_isLaoZiHao = nObject.getOrDefault("brand_isLaoZiHao", "-1").toString
        val shopName = nObject.getOrDefault("shopName", "-1").toString
        val if_state_level_new_areas = nObject.getOrDefault("if_state_level_new_areas", "-1").toString
        val city_grade = nObject.getOrDefault("city_grade", "-1").toString
        val title = nObject.getOrDefault("title", "-1").toString
        val categoryName = nObject.getOrDefault("categoryName", "-1").toString
        val firstCategoryId = nObject.getOrDefault("firstCategoryId", "-1").toString
        val thirdCategoryId = nObject.getOrDefault("thirdCategoryId", "-1").toString
        val subCategoryName = nObject.getOrDefault("subCategoryName", "-1").toString
        val salesAmount = nObject.getOrDefault("salesAmount", "-1").toString
        val administrative_region = nObject.getOrDefault("administrative_region", "-1").toString
        val platformName = nObject.getOrDefault("platformName", "-1").toString
        val priceText = nObject.getOrDefault("priceText", "-1").toString
        val if_district = nObject.getOrDefault("if_district", "-1").toString
        val brand_state = nObject.getOrDefault("brand_state", "-1").toString
        val address = nObject.getOrDefault("address", "-1").toString
        val the_yangtze_river_economic_zone_city = nObject.getOrDefault("the_yangtze_river_economic_zone_city", "-1").toString
        val originProduction = nObject.getOrDefault("originProduction", "-1").toString
        val platformId = nObject.getOrDefault("platformId", "-1").toString
        val rootCategoryId = nObject.getOrDefault("rootCategoryId", "-1").toString
        val timeStamp = nObject.getOrDefault("timeStamp", "-1").toString
        val original_cost = nObject.getOrDefault("original_cost", "-1").toString
        val the_belt_and_road_city = nObject.getOrDefault("the_belt_and_road_city", "-1").toString
        val brandValueId = nObject.getOrDefault("brandValueId", "-1").toString
        val the_belt_and_road_province = nObject.getOrDefault("the_belt_and_road_province", "-1").toString
        val originCountryName = nObject.getOrDefault("originCountryName", "-1").toString
        val rootCategoryName = nObject.getOrDefault("rootCategoryName", "-1").toString
        val regional_ID = nObject.getOrDefault("regional_ID", "-1").toString
        val brandName_cn = nObject.getOrDefault("brandName_cn", "-1").toString
        val categoryId = nObject.getOrDefault("categoryId", "-1").toString

        val model = nObject.getOrDefault("model", "-1").toString
        val color = nObject.getOrDefault("color", "-1").toString
        val ram = nObject.getOrDefault("ram", "-1").toString
        val rom = nObject.getOrDefault("rom", "-1").toString
        val series = nObject.getOrDefault("series", "-1").toString

        nObject.put("is_oversea", is_oversea)
        nObject.put("fourthCategoryId", fourthCategoryId)
        nObject.put("good_id", good_id)
        nObject.put("urban_agglomerations", urban_agglomerations)
        nObject.put("economic_division", economic_division)
        nObject.put("province", province)
        nObject.put("poor_counties", poor_counties)
        nObject.put("brandSourceId", brandSourceId)
        nObject.put("rural_demonstration_counties", rural_demonstration_counties)
        nObject.put("shopId", shopId)
        nObject.put("brand_type", brand_type)
        nObject.put("brandName", brandName)
        nObject.put("images", images)
        nObject.put("good_code", good_code)
        nObject.put("goodUrl", goodUrl)
        nObject.put("city_origin", city_origin)
        nObject.put("if_city", if_city)
        nObject.put("crawl_date", crawl_date)
        nObject.put("subCategoryId", subCategoryId)
        nObject.put("commentCount", commentCount)
        nObject.put("is_onsell", is_onsell)
        nObject.put("brandName_en", brandName_en)
        nObject.put("rural_ecommerce", rural_ecommerce)
        nObject.put("district", district)
        nObject.put("delivery_from", delivery_from)
        nObject.put("district_origin", district_origin)
        nObject.put("name", name)
        nObject.put("sellCount", sellCount)
        nObject.put("shopType", shopType)
        nObject.put("the_yangtze_river_economic_zone_province", the_yangtze_river_economic_zone_province)
        nObject.put("secondCategoryId", secondCategoryId)
        nObject.put("city", city)
        nObject.put("Base_Info", Base_Info)
        nObject.put("brand_isLaoZiHao", brand_isLaoZiHao)
        nObject.put("shopName", shopName)
        nObject.put("if_state_level_new_areas", if_state_level_new_areas)
        nObject.put("city_grade", city_grade)
        nObject.put("title", title)
        nObject.put("categoryName", categoryName)
        nObject.put("firstCategoryId", firstCategoryId)
        nObject.put("thirdCategoryId", thirdCategoryId)
        nObject.put("subCategoryName", subCategoryName)
        nObject.put("salesAmount", salesAmount)
        nObject.put("administrative_region", administrative_region)
        nObject.put("platformName", platformName)
        nObject.put("priceText", priceText)
        nObject.put("if_district", if_district)
        nObject.put("brand_state", brand_state)
        nObject.put("address", address)
        nObject.put("the_yangtze_river_economic_zone_city", the_yangtze_river_economic_zone_city)
        nObject.put("originProduction", originProduction)
        nObject.put("platformId", platformId)
        nObject.put("rootCategoryId", rootCategoryId)
        nObject.put("timeStamp", timeStamp)
        nObject.put("original_cost", original_cost)
        nObject.put("the_belt_and_road_city", the_belt_and_road_city)
        nObject.put("brandValueId", brandValueId)
        nObject.put("the_belt_and_road_province", the_belt_and_road_province)
        nObject.put("originCountryName", originCountryName)
        nObject.put("rootCategoryName", rootCategoryName)
        nObject.put("regional_ID", regional_ID)
        nObject.put("brandName_cn", brandName_cn)
        nObject.put("categoryId", categoryId)

        nObject.put("model", model)
        nObject.put("color", color)
        nObject.put("ram", ram)
        nObject.put("rom", rom)
        nObject.put("series", series)



        //2、去掉某些字段
        nObject.remove("add_to_field")
        nObject.remove("function")
        nObject.remove("shangGt1")

        nObject.toJSONString

      })

      //4、打印结果
      spark.read.json(t1).drop("shopId").registerTempTable("mryx_data")
      val addressdata = spark.read.option("header","true")
        .csv("s3a://o2o-dimension-table/address_table/address_table_2019/address_mapping_standard/2019_12/")
        .select("province","city","district","uniqueid").distinct().registerTempTable("addressdata")
      val data = spark.sql(
        """
          |select
          |t1.*,
          |concat('100',t2.uniqueid) as shopId
          |from
          |mryx_data t1
          |join
          |addressdata t2
          |on t1.province=t2.province and t1.city=t2.city and t1.district=t2.district
      """.stripMargin).dropDuplicates("good_id")
         .select("Base_Info",
        "address",
        "administrative_region",
        "brandName",
        "brandName_cn",
        "brandName_en",
        "brandSourceId",
        "brandValueId",
        "brand_isLaoZiHao",
        "brand_state",
        "brand_type",
        "categoryId",
        "categoryName",
        "city",
        "city_grade",
        "city_origin",
        "color",
        "commentCount",
        "crawl_date",
        "delivery_from",
        "district",
        "district_origin",
        "economic_division",
        "firstCategoryId",
        "fourthCategoryId",
        "goodUrl",
        "good_code",
        "good_id",
        "if_city",
        "if_district",
        "if_state_level_new_areas",
        "images",
        "is_onsell",
        "is_oversea",
        "model",
        "name",
        "originCountryName",
        "originProduction",
        "original_cost",
        "platformId",
        "platformName",
        "poor_counties",
        "priceText",
        "province",
        "ram",
        "regional_ID",
        "rom",
        "rootCategoryId",
        "rootCategoryName",
        "rural_demonstration_counties",
        "rural_ecommerce",
        "salesAmount",
        "secondCategoryId",
        "sellCount",
        "series",
        "shopId",
        "shopName",
        "shopType",
        "subCategoryId",
        "subCategoryName",
        "the_belt_and_road_city",
        "the_belt_and_road_province",
        "the_yangtze_river_economic_zone_city",
        "the_yangtze_river_economic_zone_province",
        "thirdCategoryId",
        "timeStamp",
        "title",
        "urban_agglomerations")
        .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

    }
  }
  //11、提取womaiwang
  def ES_OBS_11_womaiwang(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {
      val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
      var t1 = value.map(x => {
        val nObject = JSON.parseObject(x)

        //1、将所有的字段转为string类型

        val is_onsell  = nObject.getOrDefault("is_onsell","-1").toString
        val fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
        val good_id  = nObject.getOrDefault("good_id","-1").toString
        val urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
        val economic_division  = nObject.getOrDefault("economic_division","-1").toString
        val province  = nObject.getOrDefault("province","-1").toString
        val poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
        val brandSourceId  = nObject.getOrDefault("brandSourceId","-1").toString
        val rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
        val shopId  = nObject.getOrDefault("shopId","-1").toString
        val brand_type  = nObject.getOrDefault("brand_type","-1").toString
        val brandName  = nObject.getOrDefault("brandName","-1").toString
        val images  = nObject.getOrDefault("images","-1").toString
        val goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
        val city_origin  = nObject.getOrDefault("city_origin","-1").toString
        val if_city  = nObject.getOrDefault("if_city","-1").toString
        val crawl_date  = nObject.getOrDefault("crawl_date","-1").toString
        val subCategoryId  = nObject.getOrDefault("subCategoryId","-1").toString
        val commentCount  = nObject.getOrDefault("commentCount","-1").toString
        val brandName_en  = nObject.getOrDefault("brandName_en","-1").toString
        val rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
        val district  = nObject.getOrDefault("district","-1").toString
        val district_origin  = nObject.getOrDefault("district_origin","-1").toString
        val name  = nObject.getOrDefault("name","-1").toString
        val sellCount  = nObject.getOrDefault("sellCount","-1").toString
        val shopType  = nObject.getOrDefault("shopType","-1").toString
        val the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
        val secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
        val city  = nObject.getOrDefault("city","-1").toString
        val Base_Info  = nObject.getOrDefault("Base_Info","-1").toString
        val brand_isLaoZiHao  = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
        val shopName  = nObject.getOrDefault("shopName","-1").toString
        val if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
        val city_grade  = nObject.getOrDefault("city_grade","-1").toString
        val title  = nObject.getOrDefault("title","-1").toString
        val categoryName  = nObject.getOrDefault("categoryName","-1").toString
        val firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
        val thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
        val subCategoryName  = nObject.getOrDefault("subCategoryName","-1").toString
        val goodCount  = nObject.getOrDefault("goodCount","-1").toString
        val salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
        val administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
        val platformName  = nObject.getOrDefault("platformName","-1").toString
        val priceText  = nObject.getOrDefault("priceText","-1").toString
        val if_district  = nObject.getOrDefault("if_district","-1").toString
        val poorCount  = nObject.getOrDefault("poorCount","-1").toString
        val brand_state  = nObject.getOrDefault("brand_state","-1").toString
        val address  = nObject.getOrDefault("address","-1").toString
        val quantity  = nObject.getOrDefault("quantity","-1").toString
        val the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
        val platformId  = nObject.getOrDefault("platformId","-1").toString
        val rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
        val timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
        val generalCount  = nObject.getOrDefault("generalCount","-1").toString
        val the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
        val brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
        val the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
        val rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
        val agric_products  = nObject.getOrDefault("agric_products","-1").toString
        val regional_ID  = nObject.getOrDefault("regional_ID","-1").toString
        val brandName_cn  = nObject.getOrDefault("brandName_cn","-1").toString
        val categoryId  = nObject.getOrDefault("categoryId","-1").toString

        val model  = nObject.getOrDefault("model","-1").toString
        val color  = nObject.getOrDefault("color","-1").toString
        val ram  = nObject.getOrDefault("ram","-1").toString
        val rom  = nObject.getOrDefault("rom","-1").toString
        val series = nObject.getOrDefault("series","-1").toString
        val customCategoryId = nObject.getOrDefault("customCategoryId","-1").toString
        val original_cost = nObject.getOrDefault("original_cost","-1").toString

        nObject.put("is_onsell",is_onsell)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("good_id",good_id)
        nObject.put("urban_agglomerations",urban_agglomerations)
        nObject.put("economic_division",economic_division)
        nObject.put("province",province)
        nObject.put("poor_counties",poor_counties)
        nObject.put("brandSourceId",brandSourceId)
        nObject.put("rural_demonstration_counties",rural_demonstration_counties)
        nObject.put("shopId",shopId)
        nObject.put("brand_type",brand_type)
        nObject.put("brandName",brandName)
        nObject.put("images",images)
        nObject.put("goodUrl",goodUrl)
        nObject.put("city_origin",city_origin)
        nObject.put("if_city",if_city)
        nObject.put("crawl_date",crawl_date)
        nObject.put("subCategoryId",subCategoryId)
        nObject.put("commentCount",commentCount)
        nObject.put("brandName_en",brandName_en)
        nObject.put("rural_ecommerce",rural_ecommerce)
        nObject.put("district",district)
        nObject.put("district_origin",district_origin)
        nObject.put("name",name)
        nObject.put("sellCount",sellCount)
        nObject.put("shopType",shopType)
        nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("city",city)
        nObject.put("Base_Info",Base_Info)
        nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
        nObject.put("shopName",shopName)
        nObject.put("if_state_level_new_areas",if_state_level_new_areas)
        nObject.put("city_grade",city_grade)
        nObject.put("title",title)
        nObject.put("categoryName",categoryName)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("subCategoryName",subCategoryName)
        nObject.put("goodCount",goodCount)
        nObject.put("salesAmount",salesAmount)
        nObject.put("administrative_region",administrative_region)
        nObject.put("platformName",platformName)
        nObject.put("priceText",priceText)
        nObject.put("if_district",if_district)
        nObject.put("poorCount",poorCount)
        nObject.put("brand_state",brand_state)
        nObject.put("address",address)
        nObject.put("quantity",quantity)
        nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
        nObject.put("platformId",platformId)
        nObject.put("rootCategoryId",rootCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("generalCount",generalCount)
        nObject.put("the_belt_and_road_city",the_belt_and_road_city)
        nObject.put("brandValueId",brandValueId)
        nObject.put("the_belt_and_road_province",the_belt_and_road_province)
        nObject.put("rootCategoryName",rootCategoryName)
        nObject.put("agric_products",agric_products)
        nObject.put("regional_ID",regional_ID)
        nObject.put("brandName_cn",brandName_cn)
        nObject.put("categoryId",categoryId)
//-------------------地址

        nObject.put("address","北京市朝阳区朝阳门南大街8号")
        nObject.put("city","北京市")
        nObject.put("the_yangtze_river_economic_zone_city","0")
        nObject.put("if_city","0")
        nObject.put("city_origin","北京市")
        nObject.put("if_state_level_new_areas","0")
        nObject.put("city_grade","1")
        nObject.put("urban_agglomerations","3")
        nObject.put("economic_division","2")
        nObject.put("province","北京市")
        nObject.put("the_belt_and_road_city","0")
        nObject.put("rural_ecommerce","0")
        nObject.put("poor_counties","0")
        nObject.put("the_belt_and_road_province","0")
        nObject.put("administrative_region","华北地区")
        nObject.put("district","朝阳区")
        nObject.put("district_origin","朝阳区")
        nObject.put("shopId","80006105")
        nObject.put("the_yangtze_river_economic_zone_province","0")
        nObject.put("rural_demonstration_counties","0")
        nObject.put("regional_ID","110105")
        nObject.put("if_district","2")
//--------------------
        nObject.put("model",model)
        nObject.put("color",color)
        nObject.put("ram",ram)
        nObject.put("rom",rom)
        nObject.put("series",series)
        nObject.put("customCategoryId",customCategoryId)
        nObject.put("original_cost",original_cost)

        //2、去掉某些字段
        nObject.remove("add_to_field")
        nObject.remove("function")
        nObject.remove("shangGt1")

        nObject.toJSONString

      })

      //4、打印结果
      //      ssc.read.json(t1).show()
      spark.read.json(t1)
        .drop("shopId").registerTempTable("mryx_data")
      val addressdata = spark.read.option("header","true")
        .csv("s3a://o2o-dimension-table/address_table/address_table_2019/address_mapping_standard/2019_12/")
        .select("province","city","district","uniqueid").distinct().registerTempTable("addressdata")
      val data = spark.sql(
        """
          |select
          |t1.*,
          |concat('100',t2.uniqueid) as shopId
          |from
          |mryx_data t1
          |join
          |addressdata t2
          |on t1.province=t2.province and t1.city=t2.city and t1.district=t2.district
      """.stripMargin).dropDuplicates("good_id")
        .select("Base_Info",
          "address",
          "administrative_region",
          "agric_products",
          "brandName",
          "brandName_cn",
          "brandName_en",
          "brandSourceId",
          "brandValueId",
          "brand_isLaoZiHao",
          "brand_state",
          "brand_type",
          "categoryId",
          "categoryName",
          "city",
          "city_grade",
          "city_origin",
          "color",
          "commentCount",
          "crawl_date",
          "customCategoryId",
          "district",
          "district_origin",
          "economic_division",
          "firstCategoryId",
          "fourthCategoryId",
          "generalCount",
          "goodCount",
          "goodUrl",
          "good_id",
          "if_city",
          "if_district",
          "if_state_level_new_areas",
          "images",
          "is_onsell",
          "model",
          "name",
          "original_cost",
          "platformId",
          "platformName",
          "poorCount",
          "poor_counties",
          "priceText",
          "province",
          "quantity",
          "ram",
          "regional_ID",
          "rom",
          "rootCategoryId",
          "rootCategoryName",
          "rural_demonstration_counties",
          "rural_ecommerce",
          "salesAmount",
          "secondCategoryId",
          "sellCount",
          "series",
          "shopId",
          "shopName",
          "shopType",
          "subCategoryId",
          "subCategoryName",
          "the_belt_and_road_city",
          "the_belt_and_road_province",
          "the_yangtze_river_economic_zone_city",
          "the_yangtze_river_economic_zone_province",
          "thirdCategoryId",
          "timeStamp",
          "title",
          "urban_agglomerations")
        .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

    }
  }
  //12、提取yunji
  def ES_OBS_12_yunji(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {
      val value: RDD[String] = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values
      var t1 = value.map(x => {
        val nObject = JSON.parseObject(x)

        //1、将所有的字段转为string类型

        val country  = nObject.getOrDefault("country","-1").toString
        val is_oversea  = nObject.getOrDefault("is_oversea","-1").toString
        val fourthCategoryId  = nObject.getOrDefault("fourthCategoryId","-1").toString
        val good_id  = nObject.getOrDefault("good_id","-1").toString
        val urban_agglomerations  = nObject.getOrDefault("urban_agglomerations","-1").toString
        val economic_division  = nObject.getOrDefault("economic_division","-1").toString
        val province  = nObject.getOrDefault("province","-1").toString
        val poor_counties  = nObject.getOrDefault("poor_counties","-1").toString
        val allItemCount  = nObject.getOrDefault("allItemCount","-1").toString
        val brandSourceId  = nObject.getOrDefault("brandSourceId","-1").toString
        val goodsTaxAmount  = nObject.getOrDefault("goodsTaxAmount","-1").toString
        val rural_demonstration_counties  = nObject.getOrDefault("rural_demonstration_counties","-1").toString
        val shopId  = nObject.getOrDefault("shopId","-1").toString
        val brand_type  = nObject.getOrDefault("brand_type","-1").toString
        val isNewItem  = nObject.getOrDefault("isNewItem","-1").toString
        val brandName  = nObject.getOrDefault("brandName","-1").toString
        val images  = nObject.getOrDefault("images","-1").toString
        val goodUrl  = nObject.getOrDefault("goodUrl","-1").toString
        val city_origin  = nObject.getOrDefault("city_origin","-1").toString
        val if_city  = nObject.getOrDefault("if_city","-1").toString
        val subCategoryId  = nObject.getOrDefault("subCategoryId","-1").toString
        val shopUrl  = nObject.getOrDefault("shopUrl","-1").toString
        val serviceCommitment  = nObject.getOrDefault("serviceCommitment","-1").toString
        val brandName_en  = nObject.getOrDefault("brandName_en","-1").toString
        val rural_ecommerce  = nObject.getOrDefault("rural_ecommerce","-1").toString
        val company_name  = nObject.getOrDefault("company_name","-1").toString
        val district  = nObject.getOrDefault("district","-1").toString
        val subtitle  = nObject.getOrDefault("subtitle","-1").toString
        val taxFreeZone  = nObject.getOrDefault("taxFreeZone","-1").toString
        val delivery_from  = nObject.getOrDefault("delivery_from","-1").toString
        val district_origin  = nObject.getOrDefault("district_origin","-1").toString
        val name  = nObject.getOrDefault("name","-1").toString
        val sellCount  = nObject.getOrDefault("sellCount","-1").toString
        val shopType  = nObject.getOrDefault("shopType","-1").toString
        val the_yangtze_river_economic_zone_province  = nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
        val secondCategoryId  = nObject.getOrDefault("secondCategoryId","-1").toString
        val is_market  = nObject.getOrDefault("is_market","-1").toString
        val city  = nObject.getOrDefault("city","-1").toString
        val Base_Info  = nObject.getOrDefault("Base_Info","-1").toString
        val brand_isLaoZiHao  = nObject.getOrDefault("brand_isLaoZiHao","-1").toString
        val shopName  = nObject.getOrDefault("shopName","-1").toString
        val if_state_level_new_areas  = nObject.getOrDefault("if_state_level_new_areas","-1").toString
        val city_grade  = nObject.getOrDefault("city_grade","-1").toString
        val title  = nObject.getOrDefault("title","-1").toString
        val categoryName  = nObject.getOrDefault("categoryName","-1").toString
        val firstCategoryId  = nObject.getOrDefault("firstCategoryId","-1").toString
        val thirdCategoryId  = nObject.getOrDefault("thirdCategoryId","-1").toString
        val subCategoryName  = nObject.getOrDefault("subCategoryName","-1").toString
        val salesAmount  = nObject.getOrDefault("salesAmount","-1").toString
        val administrative_region  = nObject.getOrDefault("administrative_region","-1").toString
        val platformName  = nObject.getOrDefault("platformName","-1").toString
        val priceText  = nObject.getOrDefault("priceText","-1").toString
        val if_district  = nObject.getOrDefault("if_district","-1").toString
        val brand_state  = nObject.getOrDefault("brand_state","-1").toString
        val address  = nObject.getOrDefault("address","-1").toString
        val the_yangtze_river_economic_zone_city  = nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
        val platformId  = nObject.getOrDefault("platformId","-1").toString
        val fans  = nObject.getOrDefault("fans","-1").toString
        val rootCategoryId  = nObject.getOrDefault("rootCategoryId","-1").toString
        val timeStamp  = nObject.getOrDefault("timeStamp","-1").toString
        val original_cost  = nObject.getOrDefault("original_cost","-1").toString
        val the_belt_and_road_city  = nObject.getOrDefault("the_belt_and_road_city","-1").toString
        val brandValueId  = nObject.getOrDefault("brandValueId","-1").toString
        val the_belt_and_road_province  = nObject.getOrDefault("the_belt_and_road_province","-1").toString
        val originCountryName  = nObject.getOrDefault("originCountryName","-1").toString
        val is_selfSupport  = nObject.getOrDefault("is_selfSupport","-1").toString
        val rootCategoryName  = nObject.getOrDefault("rootCategoryName","-1").toString
        val regional_ID  = nObject.getOrDefault("regional_ID","-1").toString
        val brandName_cn  = nObject.getOrDefault("brandName_cn","-1").toString
        val categoryId  = nObject.getOrDefault("categoryId","-1").toString

        val model  = nObject.getOrDefault("model","-1").toString
        val color  = nObject.getOrDefault("color","-1").toString
        val ram  = nObject.getOrDefault("ram","-1").toString
        val rom  = nObject.getOrDefault("rom","-1").toString
        val series = nObject.getOrDefault("series","-1").toString

        nObject.put("country",country)
        nObject.put("is_oversea",is_oversea)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("good_id",good_id)
        nObject.put("urban_agglomerations",urban_agglomerations)
        nObject.put("economic_division",economic_division)
        nObject.put("province",province)
        nObject.put("poor_counties",poor_counties)
        nObject.put("allItemCount",allItemCount)
        nObject.put("brandSourceId",brandSourceId)
        nObject.put("goodsTaxAmount",goodsTaxAmount)
        nObject.put("rural_demonstration_counties",rural_demonstration_counties)
        nObject.put("shopId",shopId)
        nObject.put("brand_type",brand_type)
        nObject.put("isNewItem",isNewItem)
        nObject.put("brandName",brandName)
        nObject.put("images",images)
        nObject.put("goodUrl",goodUrl)
        nObject.put("city_origin",city_origin)
        nObject.put("if_city",if_city)
        nObject.put("subCategoryId",subCategoryId)
        nObject.put("shopUrl",shopUrl)
        nObject.put("serviceCommitment",serviceCommitment)
        nObject.put("brandName_en",brandName_en)
        nObject.put("rural_ecommerce",rural_ecommerce)
        nObject.put("company_name",company_name)
        nObject.put("district",district)
        nObject.put("subtitle",subtitle)
        nObject.put("taxFreeZone",taxFreeZone)
        nObject.put("delivery_from",delivery_from)
        nObject.put("district_origin",district_origin)
        nObject.put("name",name)
        nObject.put("sellCount",sellCount)
        nObject.put("shopType",shopType)
        nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("is_market",is_market)
        nObject.put("city",city)
        nObject.put("Base_Info",Base_Info)
        nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
        nObject.put("shopName",shopName)
        nObject.put("if_state_level_new_areas",if_state_level_new_areas)
        nObject.put("city_grade",city_grade)
        nObject.put("title",title)
        nObject.put("categoryName",categoryName)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("subCategoryName",subCategoryName)
        nObject.put("salesAmount",salesAmount)
        nObject.put("administrative_region",administrative_region)
        nObject.put("platformName",platformName)
        nObject.put("priceText",priceText)
        nObject.put("if_district",if_district)
        nObject.put("brand_state",brand_state)
        nObject.put("address",address)
        nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
        nObject.put("platformId",platformId)
        nObject.put("fans",fans)
        nObject.put("rootCategoryId",rootCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("original_cost",original_cost)
        nObject.put("the_belt_and_road_city",the_belt_and_road_city)
        nObject.put("brandValueId",brandValueId)
        nObject.put("the_belt_and_road_province",the_belt_and_road_province)
        nObject.put("originCountryName",originCountryName)
        nObject.put("is_selfSupport",is_selfSupport)
        nObject.put("rootCategoryName",rootCategoryName)
        nObject.put("regional_ID",regional_ID)
        nObject.put("brandName_cn",brandName_cn)
        nObject.put("categoryId",categoryId)

        nObject.put("model",model)
        nObject.put("color",color)
        nObject.put("ram",ram)
        nObject.put("rom",rom)
        nObject.put("series",series)


        //2、去掉某些字段
        nObject.remove("add_to_field")
        nObject.remove("function")
        nObject.remove("shangGt1")

        nObject.toJSONString

      })

      //4、打印结果
      //      ssc.read.json(t1).show()
      spark.read.json(t1).select("Base_Info",
        "address",
        "administrative_region",
        "allItemCount",
        "brandName",
        "brandName_cn",
        "brandName_en",
        "brandSourceId",
        "brandValueId",
        "brand_isLaoZiHao",
        "brand_state",
        "brand_type",
        "categoryId",
        "categoryName",
        "city",
        "city_grade",
        "city_origin",
        "color",
        "company_name",
        "country",
        "delivery_from",
        "district",
        "district_origin",
        "economic_division",
        "fans",
        "firstCategoryId",
        "fourthCategoryId",
        "goodUrl",
        "good_id",
        "goodsTaxAmount",
        "if_city",
        "if_district",
        "if_state_level_new_areas",
        "images",
        "isNewItem",
        "is_market",
        "is_oversea",
        "is_selfSupport",
        "model",
        "name",
        "originCountryName",
        "original_cost",
        "platformId",
        "platformName",
        "poor_counties",
        "priceText",
        "province",
        "ram",
        "regional_ID",
        "rom",
        "rootCategoryId",
        "rootCategoryName",
        "rural_demonstration_counties",
        "rural_ecommerce",
        "salesAmount",
        "secondCategoryId",
        "sellCount",
        "series",
        "serviceCommitment",
        "shopId",
        "shopName",
        "shopType",
        "shopUrl",
        "subCategoryId",
        "subCategoryName",
        "subtitle",
        "taxFreeZone",
        "the_belt_and_road_city",
        "the_belt_and_road_province",
        "the_yangtze_river_economic_zone_city",
        "the_yangtze_river_economic_zone_province",
        "thirdCategoryId",
        "timeStamp",
        "title",
        "urban_agglomerations")
        .repartition(4).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

    }
  }
  //13、提取dangdang
  def ES_OBS_13_dangdang(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

      val values = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values

      val frame = spark.read.json(values)

      val value = frame.toJSON.rdd.map(line => {
        val nObject = JSON.parseObject(line)

        val  is_oversea  =   nObject.getOrDefault("is_oversea","-1").toString
        val  evaluates  =   nObject.getOrDefault("evaluates","-1").toString
        val  fourthCategoryId  =   nObject.getOrDefault("fourthCategoryId","-1").toString
        val  good_id  =   nObject.getOrDefault("good_id","-1").toString
        val  urban_agglomerations  =   nObject.getOrDefault("urban_agglomerations","-1").toString
        val  economic_division  =   nObject.getOrDefault("economic_division","-1").toString
        val  score  =   nObject.getOrDefault("score","-1").toString
        val  province  =   nObject.getOrDefault("province","-1").toString
        val  poor_counties  =   nObject.getOrDefault("poor_counties","-1").toString
        val  rural_demonstration_counties  =   nObject.getOrDefault("rural_demonstration_counties","-1").toString
        val  shopId  =   nObject.getOrDefault("shopId","-1").toString
        val  brand_type  =   nObject.getOrDefault("brand_type","-1").toString
        val  goodRatePercentage  =   nObject.getOrDefault("goodRatePercentage","-1").toString
        val  brandName  =   nObject.getOrDefault("brandName","-1").toString
        val  images  =   nObject.getOrDefault("images","-1").toString
        val  favcount  =   nObject.getOrDefault("favcount","-1").toString
        val  goodUrl  =   nObject.getOrDefault("goodUrl","-1").toString
        val  city_origin  =   nObject.getOrDefault("city_origin","-1").toString
        val  if_city  =   nObject.getOrDefault("if_city","-1").toString
        val  crawl_date  =   nObject.getOrDefault("crawl_date","-1").toString
        val  subCategoryId  =   nObject.getOrDefault("subCategoryId","-1").toString
        val  shopUrl  =   nObject.getOrDefault("shopUrl","-1").toString
        val  commentCount  =   nObject.getOrDefault("commentCount","-1").toString
        val  detailCategoryName  =   nObject.getOrDefault("detailCategoryName","-1").toString
        val  brandName_en  =   nObject.getOrDefault("brandName_en","-1").toString
        val  rural_ecommerce  =   nObject.getOrDefault("rural_ecommerce","-1").toString
        val  district  =   nObject.getOrDefault("district","-1").toString
        val  taxFreeZone  =   nObject.getOrDefault("taxFreeZone","-1").toString
        val  delivery_from  =   nObject.getOrDefault("delivery_from","-1").toString
        val  district_origin  =   nObject.getOrDefault("district_origin","-1").toString
        val  name  =   nObject.getOrDefault("name","-1").toString
        val  sellCount  =   nObject.getOrDefault("sellCount","-1").toString
        val  shopType  =   nObject.getOrDefault("shopType","-1").toString
        val  the_yangtze_river_economic_zone_province  =   nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
        val  secondCategoryId  =   nObject.getOrDefault("secondCategoryId","-1").toString
        val  city  =   nObject.getOrDefault("city","-1").toString
        val  Base_Info  =   nObject.getOrDefault("Base_Info","-1").toString
        val  detailCategoryId  =   nObject.getOrDefault("detailCategoryId","-1").toString
        val  brand_isLaoZiHao  =   nObject.getOrDefault("brand_isLaoZiHao","-1").toString
        val  shopName  =   nObject.getOrDefault("shopName","-1").toString
        val  if_state_level_new_areas  =   nObject.getOrDefault("if_state_level_new_areas","-1").toString
        val  city_grade  =   nObject.getOrDefault("city_grade","-1").toString
        val  title  =   nObject.getOrDefault("title","-1").toString
        val  categoryName  =   nObject.getOrDefault("categoryName","-1").toString
        val  firstCategoryId  =   nObject.getOrDefault("firstCategoryId","-1").toString
        val  thirdCategoryId  =   nObject.getOrDefault("thirdCategoryId","-1").toString
        val  subCategoryName  =   nObject.getOrDefault("subCategoryName","-1").toString
        val  goodCount  =   nObject.getOrDefault("goodCount","-1").toString
        val  salesAmount  =   nObject.getOrDefault("salesAmount","-1").toString
        val  administrative_region  =   nObject.getOrDefault("administrative_region","-1").toString
        val  is_ebook  =   nObject.getOrDefault("is_ebook","-1").toString
        val  platformName  =   nObject.getOrDefault("platformName","-1").toString
        val  priceText  =   nObject.getOrDefault("priceText","-1").toString
        val  if_district  =   nObject.getOrDefault("if_district","-1").toString
        val  poorCount  =   nObject.getOrDefault("poorCount","-1").toString
        val  brand_state  =   nObject.getOrDefault("brand_state","-1").toString
        val  address  =   nObject.getOrDefault("address","-1").toString
        val  the_yangtze_river_economic_zone_city  =   nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
        val  platformId  =   nObject.getOrDefault("platformId","-1").toString
        val  rootCategoryId  =   nObject.getOrDefault("rootCategoryId","-1").toString
        val  timeStamp  =   nObject.getOrDefault("timeStamp","-1").toString
        val  original_cost  =   nObject.getOrDefault("original_cost","-1").toString
        val  generalCount  =   nObject.getOrDefault("generalCount","-1").toString
        val  the_belt_and_road_city  =   nObject.getOrDefault("the_belt_and_road_city","-1").toString
        val  brandValueId  =   nObject.getOrDefault("brandValueId","-1").toString
        val  originItem  =   nObject.getOrDefault("originItem","-1").toString
        val  the_belt_and_road_province  =   nObject.getOrDefault("the_belt_and_road_province","-1").toString
        val  originCountryName  =   nObject.getOrDefault("originCountryName","-1").toString
        val  is_selfSupport  =   nObject.getOrDefault("is_selfSupport","-1").toString
        val  rootCategoryName  =   nObject.getOrDefault("rootCategoryName","-1").toString
        val  regional_ID  =   nObject.getOrDefault("regional_ID","-1").toString
        val  brandName_cn  =   nObject.getOrDefault("brandName_cn","-1").toString
        val  categoryId  =   nObject.getOrDefault("categoryId","-1").toString



        nObject.put("is_oversea",is_oversea)
        nObject.put("evaluates",evaluates)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("good_id",good_id)
        nObject.put("score",score)
        nObject.put("shopId",shopId)
        nObject.put("brand_type",brand_type)
        nObject.put("goodRatePercentage",goodRatePercentage)
        nObject.put("brandName",brandName)
        nObject.put("images",images)
        nObject.put("favcount",favcount)
        nObject.put("goodUrl",goodUrl)
        nObject.put("crawl_date",crawl_date)
        nObject.put("subCategoryId",subCategoryId)
        nObject.put("shopUrl",shopUrl)
        nObject.put("commentCount",commentCount)
        nObject.put("detailCategoryName",detailCategoryName)
        nObject.put("brandName_en",brandName_en)
        nObject.put("taxFreeZone",taxFreeZone)
        nObject.put("delivery_from",delivery_from)
        nObject.put("sellCount",sellCount)
        nObject.put("shopType",shopType)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("Base_Info",Base_Info)
        nObject.put("detailCategoryId",detailCategoryId)
        nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
        nObject.put("shopName",shopName)
        nObject.put("title",title)
        nObject.put("categoryName",categoryName)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("subCategoryName",subCategoryName)
        nObject.put("goodCount",goodCount)
        nObject.put("salesAmount",salesAmount)
        nObject.put("is_ebook",is_ebook)
        nObject.put("platformName",platformName)
        nObject.put("priceText",priceText)
        nObject.put("poorCount",poorCount)
        nObject.put("brand_state",brand_state)
        nObject.put("platformId",platformId)
        nObject.put("rootCategoryId",rootCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("original_cost",original_cost)
        nObject.put("generalCount",generalCount)
        nObject.put("brandValueId",brandValueId)
        nObject.put("originItem",originItem)
        nObject.put("originCountryName",originCountryName)
        nObject.put("is_selfSupport",is_selfSupport)
        nObject.put("rootCategoryName",rootCategoryName)
        nObject.put("brandName_cn",brandName_cn)
        nObject.put("categoryId",categoryId)


        nObject.put("name", name)
        nObject.put("address", address)
        nObject.put("province", province)
        nObject.put("city", city)
        nObject.put("district", district)
        nObject.put("administrative_region", administrative_region)
        nObject.put("city_grade", city_grade)
        nObject.put("city_origin", city_origin)
        nObject.put("economic_division", economic_division)
        nObject.put("district_origin", district_origin)
        nObject.put("if_city", if_city)
        nObject.put("if_district",if_district)
        nObject.put("if_state_level_new_areas", if_state_level_new_areas)
        nObject.put("poor_counties", poor_counties)
        nObject.put("regional_ID",regional_ID)
        nObject.put("rural_demonstration_counties", rural_demonstration_counties)
        nObject.put("rural_ecommerce", rural_ecommerce)
        nObject.put("the_belt_and_road_city", the_belt_and_road_city)
        nObject.put("the_belt_and_road_province",the_belt_and_road_province)
        nObject.put("the_yangtze_river_economic_zone_city",the_yangtze_river_economic_zone_city)
        nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
        nObject.put("urban_agglomerations", urban_agglomerations)



        nObject.remove("add_to_field")
        nObject.remove("function")
        nObject.remove("shangGt1")

        nObject.toJSONString
      })

      spark.read.json(value).registerTempTable("frame")

      val data = spark.sql(
        """
          |select
          |
          |is_oversea,
          |evaluates,
          |fourthCategoryId,
          |good_id,
          |urban_agglomerations,
          |economic_division,
          |score,
          |province,
          |poor_counties,
          |rural_demonstration_counties,
          |shopId,
          |brand_type,
          |goodRatePercentage,
          |brandName,
          |images,
          |favcount,
          |goodUrl,
          |city_origin,
          |if_city,
          |crawl_date,
          |subCategoryId,
          |shopUrl,
          |commentCount,
          |detailCategoryName,
          |brandName_en,
          |rural_ecommerce,
          |district,
          |taxFreeZone,
          |delivery_from,
          |district_origin,
          |name,
          |sellCount,
          |shopType,
          |the_yangtze_river_economic_zone_province,
          |secondCategoryId,
          |city,
          |Base_Info,
          |detailCategoryId,
          |brand_isLaoZiHao,
          |shopName,
          |if_state_level_new_areas,
          |city_grade,
          |title,
          |categoryName,
          |firstCategoryId,
          |thirdCategoryId,
          |subCategoryName,
          |goodCount,
          |salesAmount,
          |administrative_region,
          |is_ebook,
          |platformName,
          |priceText,
          |if_district,
          |poorCount,
          |brand_state,
          |address,
          |the_yangtze_river_economic_zone_city,
          |platformId,
          |rootCategoryId,
          |timeStamp,
          |original_cost,
          |generalCount,
          |the_belt_and_road_city,
          |brandValueId,
          |originItem,
          |the_belt_and_road_province,
          |originCountryName,
          |is_selfSupport,
          |rootCategoryName,
          |regional_ID,
          |brandName_cn,
          |categoryId
          |
          |from
          |frame
        """.stripMargin)

      data.repartition(5).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")
//      data.show()
    }
  }
  //14、提取guomei
  def ES_OBS_14_guomei(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

      val values = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values

      val frame = spark.read.json(values)

      val value = frame.toJSON.rdd.map(line => {
        val nObject = JSON.parseObject(line)

        val  is_oversea                                         =   nObject.getOrDefault("is_oversea","-1").toString
        val  fourthCategoryId                                   =   nObject.getOrDefault("fourthCategoryId","-1").toString
        val  good_id                                            =   nObject.getOrDefault("good_id","-1").toString
        val  urban_agglomerations                               =   nObject.getOrDefault("urban_agglomerations","-1").toString
        val  economic_division                                  =   nObject.getOrDefault("economic_division","-1").toString
        val  province                                           =   nObject.getOrDefault("province","-1").toString
        val  poor_counties                                      =   nObject.getOrDefault("poor_counties","-1").toString
        val  brandSourceId                                      =   nObject.getOrDefault("brandSourceId","-1").toString
        val  rural_demonstration_counties                       =   nObject.getOrDefault("rural_demonstration_counties","-1").toString
        val  shopId                                             =   nObject.getOrDefault("shopId","-1").toString
        val  brand_type                                         =   nObject.getOrDefault("brand_type","-1").toString
        val  goodRatePercentage                                 =   nObject.getOrDefault("goodRatePercentage","-1").toString
        val  brandName                                          =   nObject.getOrDefault("brandName","-1").toString
        val  images                                             =   nObject.getOrDefault("images","-1").toString
        val  good_code                                          =   nObject.getOrDefault("good_code","-1").toString
        val  goodUrl                                            =   nObject.getOrDefault("goodUrl","-1").toString
        val  city_origin                                        =   nObject.getOrDefault("city_origin","-1").toString
        val  if_city                                            =   nObject.getOrDefault("if_city","-1").toString
        val  crawl_date                                         =   nObject.getOrDefault("crawl_date","-1").toString
        val  subCategoryId                                      =   nObject.getOrDefault("subCategoryId","-1").toString
        val  shopUrl                                            =   nObject.getOrDefault("shopUrl","-1").toString
        val  commentCount                                       =   nObject.getOrDefault("commentCount","-1").toString
        val  brandName_en                                       =   nObject.getOrDefault("brandName_en","-1").toString
        val  rural_ecommerce                                    =   nObject.getOrDefault("rural_ecommerce","-1").toString
        val  district                                           =   nObject.getOrDefault("district","-1").toString
        val  delivery_from                                      =   nObject.getOrDefault("delivery_from","-1").toString
        val  district_origin                                    =   nObject.getOrDefault("district_origin","-1").toString
        val  name                                               =   nObject.getOrDefault("name","-1").toString
        val  sellCount                                          =   nObject.getOrDefault("sellCount","-1").toString
        val  is_o2o                                             =   nObject.getOrDefault("is_o2o","-1").toString
        val  shopType                                           =   nObject.getOrDefault("shopType","-1").toString
        val  the_yangtze_river_economic_zone_province           =   nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
        val  secondCategoryId                                   =   nObject.getOrDefault("secondCategoryId","-1").toString
        val  city                                               =   nObject.getOrDefault("city","-1").toString
        val  Base_Info                                          =   nObject.getOrDefault("Base_Info","-1").toString
        val  brand_isLaoZiHao                                   =   nObject.getOrDefault("brand_isLaoZiHao","-1").toString
        val  shopName                                           =   nObject.getOrDefault("shopName","-1").toString
        val  if_state_level_new_areas                           =   nObject.getOrDefault("if_state_level_new_areas","-1").toString
        val  city_grade                                         =   nObject.getOrDefault("city_grade","-1").toString
        val  title                                              =   nObject.getOrDefault("title","-1").toString
        val  categoryName                                       =   nObject.getOrDefault("categoryName","-1").toString
        val  firstCategoryId                                    =   nObject.getOrDefault("firstCategoryId","-1").toString
        val  thirdCategoryId                                    =   nObject.getOrDefault("thirdCategoryId","-1").toString
        val  subCategoryName                                    =   nObject.getOrDefault("subCategoryName","-1").toString
        val  goodCount                                          =   nObject.getOrDefault("goodCount","-1").toString
        val  salesAmount                                        =   nObject.getOrDefault("salesAmount","-1").toString
        val  administrative_region                              =   nObject.getOrDefault("administrative_region","-1").toString
        val  platformName                                       =   nObject.getOrDefault("platformName","-1").toString
        val  priceText                                          =   nObject.getOrDefault("priceText","-1").toString
        val  if_district                                        =   nObject.getOrDefault("if_district","-1").toString
        val  poorCount                                          =   nObject.getOrDefault("poorCount","-1").toString
        val  brand_state                                        =   nObject.getOrDefault("brand_state","-1").toString
        val  address                                            =   nObject.getOrDefault("address","-1").toString
        val  the_yangtze_river_economic_zone_city               =   nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
        val  platformId                                         =   nObject.getOrDefault("platformId","-1").toString
        val  promotion_price                                    =   nObject.getOrDefault("promotion_price","-1").toString
        val  rootCategoryId                                     =   nObject.getOrDefault("rootCategoryId","-1").toString
        val  timeStamp                                          =   nObject.getOrDefault("timeStamp","-1").toString
        val  shop_good_id                                       =   nObject.getOrDefault("shop_good_id","-1").toString
        val  original_cost                                      =   nObject.getOrDefault("original_cost","-1").toString
        val  generalCount                                       =   nObject.getOrDefault("generalCount","-1").toString
        val  the_belt_and_road_city                             =   nObject.getOrDefault("the_belt_and_road_city","-1").toString
        val  brandValueId                                       =   nObject.getOrDefault("brandValueId","-1").toString
        val  the_belt_and_road_province                         =   nObject.getOrDefault("the_belt_and_road_province","-1").toString
        val  originCountryName                                  =   nObject.getOrDefault("originCountryName","-1").toString
        val  is_selfSupport                                     =   nObject.getOrDefault("is_selfSupport","-1").toString
        val  rootCategoryName                                   =   nObject.getOrDefault("rootCategoryName","-1").toString
        val  regional_ID                                        =   nObject.getOrDefault("regional_ID","-1").toString
        val  brandName_cn                                       =   nObject.getOrDefault("brandName_cn","-1").toString
        val  categoryId  =   nObject.getOrDefault("categoryId","-1").toString
        nObject.put("is_oversea",is_oversea)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("good_id",good_id)
        nObject.put("brandSourceId",brandSourceId)
        nObject.put("shopId",shopId)
        nObject.put("brand_type",brand_type)
        nObject.put("goodRatePercentage",goodRatePercentage)
        nObject.put("brandName",brandName)
        nObject.put("images",images)
        nObject.put("good_code",good_code)
        nObject.put("goodUrl",goodUrl)
        nObject.put("crawl_date",crawl_date)
        nObject.put("subCategoryId",subCategoryId)
        nObject.put("shopUrl",shopUrl)
        nObject.put("commentCount",commentCount)
        nObject.put("brandName_en",brandName_en)
        nObject.put("delivery_from",delivery_from)
        nObject.put("sellCount",sellCount)
        nObject.put("is_o2o",is_o2o)
        nObject.put("shopType",shopType)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("Base_Info",Base_Info)
        nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
        nObject.put("shopName",shopName)
        nObject.put("title",title)
        nObject.put("categoryName",categoryName)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("subCategoryName",subCategoryName)
        nObject.put("goodCount",goodCount)
        nObject.put("salesAmount",salesAmount)
        nObject.put("platformName",platformName)
        nObject.put("priceText",priceText)
        nObject.put("poorCount",poorCount)
        nObject.put("brand_state",brand_state)
        nObject.put("platformId",platformId)
        nObject.put("promotion_price",promotion_price)
        nObject.put("rootCategoryId",rootCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("shop_good_id",shop_good_id)
        nObject.put("original_cost",original_cost)
        nObject.put("generalCount",generalCount)
        nObject.put("brandValueId",brandValueId)
        nObject.put("originCountryName",originCountryName)
        nObject.put("is_selfSupport",is_selfSupport)
        nObject.put("rootCategoryName",rootCategoryName)
        nObject.put("brandName_cn",brandName_cn)
        nObject.put("categoryId",categoryId)



        nObject.put("regional_ID", regional_ID)
        nObject.put("name", name)
        nObject.put("address", address)
        nObject.put("administrative_region", administrative_region)
        nObject.put("province", province)
        nObject.put("city_origin", city_origin)
        nObject.put("district_origin", district_origin)
        nObject.put("city", city)
        nObject.put("district", district)
        nObject.put("city_grade",city_grade)
        nObject.put("urban_agglomerations", urban_agglomerations)
        nObject.put("economic_division", economic_division)
        nObject.put("if_city", if_city)
        nObject.put("if_district", if_district)
        nObject.put("poor_counties", poor_counties)
        nObject.put("rural_demonstration_counties", rural_demonstration_counties)
        nObject.put("rural_ecommerce",rural_ecommerce)
        nObject.put("if_state_level_new_areas", if_state_level_new_areas)
        nObject.put("the_yangtze_river_economic_zone_province",the_yangtze_river_economic_zone_province)
        nObject.put("the_yangtze_river_economic_zone_city", the_yangtze_river_economic_zone_city)
        nObject.put("the_belt_and_road_province", the_belt_and_road_province)
        nObject.put("the_belt_and_road_city", the_belt_and_road_city)



        nObject.remove("add_to_field")
        nObject.remove("function")
        nObject.remove("shangGt1")

        nObject.toJSONString
      })

      spark.read.json(value).registerTempTable("frame")

      val data = spark.sql(
        """
          |select
          |is_oversea,
          |fourthCategoryId,
          |good_id,
          |urban_agglomerations,
          |economic_division,
          |province,
          |poor_counties,
          |brandSourceId,
          |rural_demonstration_counties,
          |shopId,
          |brand_type,
          |goodRatePercentage,
          |brandName,
          |images,
          |good_code,
          |goodUrl,
          |city_origin,
          |if_city,
          |crawl_date,
          |subCategoryId,
          |shopUrl,
          |commentCount,
          |brandName_en,
          |rural_ecommerce,
          |district,
          |delivery_from,
          |district_origin,
          |name,
          |sellCount,
          |is_o2o,
          |shopType,
          |the_yangtze_river_economic_zone_province,
          |secondCategoryId,
          |city,
          |Base_Info,
          |brand_isLaoZiHao,
          |shopName,
          |if_state_level_new_areas,
          |city_grade,
          |title,
          |categoryName,
          |firstCategoryId,
          |thirdCategoryId,
          |subCategoryName,
          |goodCount,
          |salesAmount,
          |administrative_region,
          |platformName,
          |priceText,
          |if_district,
          |poorCount,
          |brand_state,
          |address,
          |the_yangtze_river_economic_zone_city,
          |platformId,
          |promotion_price,
          |rootCategoryId,
          |timeStamp,
          |shop_good_id,
          |original_cost,
          |generalCount,
          |the_belt_and_road_city,
          |brandValueId,
          |the_belt_and_road_province,
          |originCountryName,
          |is_selfSupport,
          |rootCategoryName,
          |regional_ID,
          |brandName_cn,
          |categoryId
          |from
          |frame
        """.stripMargin)


      data.repartition(5).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

    }
  }
  //15、提取jumei
  def ES_OBS_15_jumei(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

      val values = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values

      val frame = spark.read.json(values)

      val value = frame.toJSON.rdd.map(line => {
        val nObject = JSON.parseObject(line)

        val  is_oversea   =  nObject.getOrDefault("is_oversea","-1").toString
        val  fourthCategoryId   =  nObject.getOrDefault("fourthCategoryId","-1").toString
        val  good_id   =  nObject.getOrDefault("good_id","-1").toString
        val  urban_agglomerations   =  nObject.getOrDefault("urban_agglomerations","-1").toString
        val  economic_division   =  nObject.getOrDefault("economic_division","-1").toString
        val  province   =  nObject.getOrDefault("province","-1").toString
        val  poor_counties   =  nObject.getOrDefault("poor_counties","-1").toString
        val  brandSourceId   =  nObject.getOrDefault("brandSourceId","-1").toString
        val  rural_demonstration_counties   =  nObject.getOrDefault("rural_demonstration_counties","-1").toString
        val  shopId   =  nObject.getOrDefault("shopId","-1").toString
        val  brand_type   =  nObject.getOrDefault("brand_type","-1").toString
        val  brandName   =  nObject.getOrDefault("brandName","-1").toString
        val  images   =  nObject.getOrDefault("images","-1").toString
        val  favcount   =  nObject.getOrDefault("favcount","-1").toString
        val  goodUrl   =  nObject.getOrDefault("goodUrl","-1").toString
        val  temp_good_id   =  nObject.getOrDefault("temp_good_id","-1").toString
        val  city_origin   =  nObject.getOrDefault("city_origin","-1").toString
        val  if_city   =  nObject.getOrDefault("if_city","-1").toString
        val  subCategoryId   =  nObject.getOrDefault("subCategoryId","-1").toString
        val  shopUrl   =  nObject.getOrDefault("shopUrl","-1").toString
        val  is_onsell   =  nObject.getOrDefault("is_onsell","-1").toString
        val  start_time   =  nObject.getOrDefault("start_time","-1").toString
        val  brandName_en   =  nObject.getOrDefault("brandName_en","-1").toString
        val  rural_ecommerce   =  nObject.getOrDefault("rural_ecommerce","-1").toString
        val  district   =  nObject.getOrDefault("district","-1").toString
        val  district_origin   =  nObject.getOrDefault("district_origin","-1").toString
        val  name   =  nObject.getOrDefault("name","-1").toString
        val  sellCount   =  nObject.getOrDefault("sellCount","-1").toString
        val  shopType   =  nObject.getOrDefault("shopType","-1").toString
        val  the_yangtze_river_economic_zone_province   =  nObject.getOrDefault("the_yangtze_river_economic_zone_province","-1").toString
        val  secondCategoryId   =  nObject.getOrDefault("secondCategoryId","-1").toString
        val  city   =  nObject.getOrDefault("city","-1").toString
        val  Base_Info   =  nObject.getOrDefault("Base_Info","-1").toString
        val  brand_isLaoZiHao   =  nObject.getOrDefault("brand_isLaoZiHao","-1").toString
        val  shopName   =  nObject.getOrDefault("shopName","-1").toString
        val  kaifaqu   =  nObject.getOrDefault("kaifaqu","-1").toString
        val  if_state_level_new_areas   =  nObject.getOrDefault("if_state_level_new_areas","-1").toString
        val  city_grade   =  nObject.getOrDefault("city_grade","-1").toString
        val  title   =  nObject.getOrDefault("title","-1").toString
        val  categoryName   =  nObject.getOrDefault("categoryName","-1").toString
        val  firstCategoryId   =  nObject.getOrDefault("firstCategoryId","-1").toString
        val  thirdCategoryId   =  nObject.getOrDefault("thirdCategoryId","-1").toString
        val  subCategoryName   =  nObject.getOrDefault("subCategoryName","-1").toString
        val  salesAmount   =  nObject.getOrDefault("salesAmount","-1").toString
        val  administrative_region   =  nObject.getOrDefault("administrative_region","-1").toString
        val  platformName   =  nObject.getOrDefault("platformName","-1").toString
        val  priceText   =  nObject.getOrDefault("priceText","-1").toString
        val  if_district   =  nObject.getOrDefault("if_district","-1").toString
        val  brand_state   =  nObject.getOrDefault("brand_state","-1").toString
        val  address   =  nObject.getOrDefault("address","-1").toString
        val  item_id   =  nObject.getOrDefault("item_id","-1").toString
        val  the_yangtze_river_economic_zone_city   =  nObject.getOrDefault("the_yangtze_river_economic_zone_city","-1").toString
        val  end_time   =  nObject.getOrDefault("end_time","-1").toString
        val  good_type   =  nObject.getOrDefault("good_type","-1").toString
        val  platformId   =  nObject.getOrDefault("platformId","-1").toString
        val  fans   =  nObject.getOrDefault("fans","-1").toString
        val  rootCategoryId   =  nObject.getOrDefault("rootCategoryId","-1").toString
        val  timeStamp   =  nObject.getOrDefault("timeStamp","-1").toString
        val  original_cost   =  nObject.getOrDefault("original_cost","-1").toString
        val  the_belt_and_road_city   =  nObject.getOrDefault("the_belt_and_road_city","-1").toString
        val  brandValueId   =  nObject.getOrDefault("brandValueId","-1").toString
        val  the_belt_and_road_province   =  nObject.getOrDefault("the_belt_and_road_province","-1").toString
        val  originCountryName   =  nObject.getOrDefault("originCountryName","-1").toString
        val  is_selfSupport   =  nObject.getOrDefault("is_selfSupport","-1").toString
        val  rootCategoryName   =  nObject.getOrDefault("rootCategoryName","-1").toString
        val  regional_ID   =  nObject.getOrDefault("regional_ID","-1").toString
        val  brandName_cn   =  nObject.getOrDefault("brandName_cn","-1").toString
        val  categoryId   =  nObject.getOrDefault("categoryId","-1").toString

        nObject.put("is_oversea",is_oversea)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("good_id",good_id)
        nObject.put("brandSourceId",brandSourceId)
        nObject.put("shopId",shopId)
        nObject.put("brand_type",brand_type)
        nObject.put("brandName",brandName)
        nObject.put("images",images)
        nObject.put("favcount",favcount)
        nObject.put("goodUrl",goodUrl)
        nObject.put("temp_good_id",temp_good_id)
        nObject.put("subCategoryId",subCategoryId)
        nObject.put("shopUrl",shopUrl)
        nObject.put("is_onsell",is_onsell)
        nObject.put("start_time",start_time)
        nObject.put("brandName_en",brandName_en)
        nObject.put("sellCount",sellCount)
        nObject.put("shopType",shopType)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("Base_Info",Base_Info)
        nObject.put("brand_isLaoZiHao",brand_isLaoZiHao)
        nObject.put("shopName",shopName)
        nObject.put("title",title)
        nObject.put("categoryName",categoryName)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("subCategoryName",subCategoryName)
        nObject.put("salesAmount",salesAmount)
        nObject.put("platformName",platformName)
        nObject.put("priceText",priceText)
        nObject.put("brand_state",brand_state)
        nObject.put("item_id",item_id)
        nObject.put("end_time",end_time)
        nObject.put("good_type",good_type)
        nObject.put("platformId",platformId)
        nObject.put("fans",fans)
        nObject.put("rootCategoryId",rootCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("original_cost",original_cost)
        nObject.put("brandValueId",brandValueId)
        nObject.put("originCountryName",originCountryName)
        nObject.put("is_selfSupport",is_selfSupport)
        nObject.put("rootCategoryName",rootCategoryName)
        nObject.put("brandName_cn",brandName_cn)
        nObject.put("categoryId",categoryId)


        nObject.put("address", address)
        nObject.put("administrative_region", administrative_region)
        nObject.put("city", city)
        nObject.put("city_grade", city_grade)
        nObject.put("city_origin", city_origin)
        nObject.put("district", district)
        nObject.put("economic_division", economic_division)
        nObject.put("district_origin", district_origin)
        nObject.put("if_city", if_city)
        nObject.put("if_district", if_district)
        nObject.put("if_state_level_new_areas",if_state_level_new_areas)
        nObject.put("kaifaqu", "0")
        nObject.put("name", "北京创锐文化传媒有限公司")
        nObject.put("poor_counties", "1")
        nObject.put("province", "北京市")
        nObject.put("regional_ID", "110105")
        nObject.put("rural_demonstration_counties", "0")
        nObject.put("rural_ecommerce", "0")
        nObject.put("the_belt_and_road_city", "0")
        nObject.put("the_yangtze_river_economic_zone_city", "0")
        nObject.put("the_belt_and_road_province", "0")
        nObject.put("the_yangtze_river_economic_zone_province", "0")
        nObject.put("urban_agglomerations", "3")


        nObject.toJSONString
      })

      spark.read.json(value).registerTempTable("frame")

      val data = spark.sql(
        """
          |select
          |is_oversea,
          |fourthCategoryId,
          |good_id,
          |urban_agglomerations,
          |economic_division,
          |province,
          |poor_counties,
          |brandSourceId,
          |rural_demonstration_counties,
          |shopId,
          |brand_type,
          |brandName,
          |images,
          |favcount,
          |goodUrl,
          |temp_good_id,
          |city_origin,
          |if_city,
          |subCategoryId,
          |shopUrl,
          |is_onsell,
          |start_time,
          |brandName_en,
          |rural_ecommerce,
          |district,
          |district_origin,
          |name,
          |sellCount,
          |shopType,
          |the_yangtze_river_economic_zone_province,
          |secondCategoryId,
          |city,
          |Base_Info,
          |brand_isLaoZiHao,
          |shopName,
          |kaifaqu,
          |if_state_level_new_areas,
          |city_grade,
          |title,
          |categoryName,
          |firstCategoryId,
          |thirdCategoryId,
          |subCategoryName,
          |salesAmount,
          |administrative_region,
          |platformName,
          |priceText,
          |if_district,
          |brand_state,
          |address,
          |item_id,
          |the_yangtze_river_economic_zone_city,
          |end_time,
          |good_type,
          |platformId,
          |fans,
          |rootCategoryId,
          |timeStamp,
          |original_cost,
          |the_belt_and_road_city,
          |brandValueId,
          |the_belt_and_road_province,
          |originCountryName,
          |is_selfSupport,
          |rootCategoryName,
          |regional_ID,
          |brandName_cn,
          |categoryId
          |from
          |frame
        """.stripMargin)


      data.repartition(5).write.orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

    }
  }
  //16、提取tangeche
  def ES_OBS_16_tangeche(spark:SparkSession, year: String,platform:String,monthArr:Array[Int]): Unit = {

    val sc = spark.sparkContext
    for (month <- monthArr) {

      val values = sc.esJsonRDD(s"${year}_${platform}/${platform}_${year}_${month}").values.map(line => {
        val nObject = JSON.parseObject(line)
        val   brandName  =   nObject.getOrDefault("brandName","-1").toString
        val   images  =   nObject.getOrDefault("images","-1").toString
        val   goodUrl  =   nObject.getOrDefault("goodUrl","-1").toString
        val   seriesName  =   nObject.getOrDefault("seriesName","-1").toString
        val   good_id  =   nObject.getOrDefault("good_id","-1").toString
        val   fourthCategoryId  =   nObject.getOrDefault("fourthCategoryId","-1").toString
        val   platformId  =   nObject.getOrDefault("platformId","-1").toString
        val   title  =   nObject.getOrDefault("title","-1").toString
        val   firstCategoryId  =   nObject.getOrDefault("firstCategoryId","-1").toString
        val   seriesId  =   nObject.getOrDefault("seriesId","-1").toString
        val   thirdCategoryId  =   nObject.getOrDefault("thirdCategoryId","-1").toString
        val   timeStamp  =   nObject.getOrDefault("timeStamp","-1").toString
        val   Base_Info  =   nObject.getOrDefault("Base_Info","-1").toString
        val   specName  =   nObject.getOrDefault("specName","-1").toString
        val   brandValueId  =   nObject.getOrDefault("brandValueId","-1").toString
        val   salesAmount  =   nObject.getOrDefault("salesAmount","-1").toString
        val   sellCount  =   nObject.getOrDefault("sellCount","-1").toString
        val   guidePrice  =   nObject.getOrDefault("guidePrice","-1").toString
        val   shopType  =   nObject.getOrDefault("shopType","-1").toString
        val   platformName  =   nObject.getOrDefault("platformName","-1").toString
        val   secondCategoryId  =   nObject.getOrDefault("secondCategoryId","-1").toString
        val   priceText  =   nObject.getOrDefault("priceText","-1").toString
        val   platformName_spelling  =   nObject.getOrDefault("platformName_spelling","-1").toString


        nObject.put("brandName",brandName)
        nObject.put("images",images)
        nObject.put("goodUrl",goodUrl)
        nObject.put("seriesName",seriesName)
        nObject.put("good_id",good_id)
        nObject.put("fourthCategoryId",fourthCategoryId)
        nObject.put("platformId",platformId)
        nObject.put("title",title)
        nObject.put("firstCategoryId",firstCategoryId)
        nObject.put("seriesId",seriesId)
        nObject.put("thirdCategoryId",thirdCategoryId)
        nObject.put("timeStamp",timeStamp)
        nObject.put("Base_Info",Base_Info)
        nObject.put("specName",specName)
        nObject.put("brandValueId",brandValueId)
        nObject.put("salesAmount",salesAmount)
        nObject.put("sellCount",sellCount)
        nObject.put("guidePrice",guidePrice)
        nObject.put("shopType",shopType)
        nObject.put("platformName",platformName)
        nObject.put("secondCategoryId",secondCategoryId)
        nObject.put("priceText",priceText)
        nObject.put("platformName_spelling",platformName_spelling)

        nObject.put("name", "北京搜车网科技有限公司")
        nObject.put("address", "北京市海淀区巨山路78号院209室")
        nObject.put("province", "北京市")
        nObject.put("city", "北京市")
        nObject.put("district", "海淀区")
        nObject.put("administrative_region", "华北地区")
        nObject.put("city_grade", "1")
        nObject.put("city_origin", "北京市")
        nObject.put("economic_division", "2")
        nObject.put("district_origin", "海淀区")
        nObject.put("if_city", "0")
        nObject.put("if_district", "2")
        nObject.put("if_state_level_new_areas", "0")
        nObject.put("poor_counties", "1")
        nObject.put("regional_ID", "110108")
        nObject.put("rural_demonstration_counties", "0")
        nObject.put("rural_ecommerce", "0")
        nObject.put("the_belt_and_road_city", "0")
        nObject.put("the_belt_and_road_province", "0")
        nObject.put("the_yangtze_river_economic_zone_city", "0")
        nObject.put("the_yangtze_river_economic_zone_province", "0")
        nObject.put("urban_agglomerations", "3")
        nObject.put("longitude", "-1")
        nObject.put("latitude", "-1")


        nObject.remove("add_to_field")
        nObject.remove("function")
        nObject.remove("shangGt1")

        nObject.toJSONString
      })

      spark.read.json(values).drop("shopId").registerTempTable("mryx_data")
      val addressdata = spark.read.option("header","true")
        .csv("s3a://o2o-dimension-table/address_table/address_table_2019/address_mapping_standard/2019_12/")
        .select("province","city","district","uniqueid").distinct().registerTempTable("addressdata")
      val data = spark.sql(
        """
          |select
          |t1.*,
          |concat('100',t2.uniqueid) as shopId
          |from
          |mryx_data t1
          |join
          |addressdata t2
          |on t1.province=t2.province and t1.city=t2.city and t1.district=t2.district
      """.stripMargin).dropDuplicates("good_id").registerTempTable("frame")

      spark.sql(
        """
          |select
          |
          |brandName,
          |images,
          |goodUrl,
          |seriesName,
          |good_id,
          |fourthCategoryId,
          |platformId,
          |title,
          |firstCategoryId,
          |seriesId,
          |thirdCategoryId,
          |timeStamp,
          |Base_info,
          |specName,
          |brandValueId,
          |salesAmount,
          |sellCount,
          |guidePrice,
          |shopId,
          |shopType,
          |platformName,
          |secondCategoryId,
          |priceText,
          |platformName_spelling,
          |name,
          |address,
          |province,
          |city,
          |district,
          |administrative_region,
          |city_grade,
          |city_origin,
          |economic_division,
          |district_origin,
          |if_city,
          |if_district,
          |if_state_level_new_areas,
          |poor_counties,
          |regional_ID,
          |rural_demonstration_counties,
          |rural_ecommerce,
          |the_belt_and_road_city,
          |the_belt_and_road_province,
          |the_yangtze_river_economic_zone_city,
          |the_yangtze_river_economic_zone_province,
          |urban_agglomerations,
          |longitude,
          |latitude
          |
          |
          |from
          |frame
        """.stripMargin)
        .repartition(5).write.mode("overwrite").orc(s"s3a://dws-data/split/split_data/${year}/${platform}/${month}")

    }
  }

}
